1 /** @file
2
3 A brief file description
4
5 @section license License
6
7 Licensed to the Apache Software Foundation (ASF) under one
8 or more contributor license agreements. See the NOTICE file
9 distributed with this work for additional information
10 regarding copyright ownership. The ASF licenses this file
11 to you under the Apache License, Version 2.0 (the
12 "License"); you may not use this file except in compliance
13 with the License. You may obtain a copy of the License at
14
15 http://www.apache.org/licenses/LICENSE-2.0
16
17 Unless required by applicable law or agreed to in writing, software
18 distributed under the License is distributed on an "AS IS" BASIS,
19 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
20 See the License for the specific language governing permissions and
21 limitations under the License.
22 */
23
24 #include "I_EventSystem.h"
25 #include "I_Net.h"
26 #include "I_Cache.h"
27 #include "tscore/I_Version.h"
28
29 #include "ts/ts.h" // plugin header
30 #include "protocol_binary.h"
31 #include "tscore/ink_memory.h"
32 #include "tscore/ink_hrtime.h"
33 #include "tscore/CryptoHash.h"
34
35 #define TSMEMCACHE_VERSION "1.0.0"
36 #define TSMEMCACHE_MAX_CMD_SIZE (128 * 1024 * 1024) // silly large
37 #define TSMEMCACHE_MAX_KEY_LEN 250
38 #define TSMEMCACHE_TMP_CMD_BUFFER_SIZE 320
39 #define TSMEMCACHE_HEADER_MAGIC 0x8765ACDC
40 #define TSMEMCACHE_RETRY_WRITE_INTERVAL HRTIME_MSECONDS(20)
41
42 #define TSMEMCACHE_WRITE_SYNC 0 // not yet
43
44 #define TSMEMCACHE_EVENT_GOT_ITEM 100000
45 #define TSMEMCACHE_EVENT_GOT_KEY 100001
46 #define TSMEMCACHE_STREAM_DONE 100002
47 #define TSMEMCACHE_TUNNEL_DONE 100003
48
49 #define CHECK_RET(_e, _r) \
50 do { \
51 int ret = (_e); \
52 if (ret != _r) \
53 return _r; \
54 } while (0)
55 #define WRITE(_s) write(_s "", sizeof(_s "") - 1)
56 #define STRLEN(_s) (sizeof(_s "") - 1)
57
58 struct MCCacheHeader {
59 uint32_t magic;
60 uint32_t flags;
61 uint32_t nkey : 8;
62 uint32_t reserved : 24;
63 uint32_t exptime; // seconds offset from settime
64 uint64_t settime;
65 uint64_t cas;
66 uint64_t nbytes;
67 char *
keyMCCacheHeader68 key()
69 {
70 return ((char *)this) + sizeof(MCCacheHeader);
71 }
72 int
lenMCCacheHeader73 len()
74 {
75 return sizeof(MCCacheHeader) + nkey;
76 }
77 };
78
79 struct MCAccept : public Continuation {
80 #ifndef HAVE_TLS
81 ProxyAllocator *theMCThreadAllocator;
82 #endif
83 int accept_port = 0;
84 int main_event(int event, void *netvc);
85
MCAcceptMCAccept86 MCAccept()
87 :
88 #ifndef HAVE_TLS
89 theMCThreadAllocator(NULL)
90 #endif
91
92 {
93 SET_HANDLER(&MCAccept::main_event);
94 }
95 };
96
97 #define TS_PUSH_HANDLER(_h) \
98 do { \
99 handler_stack[ihandler_stack++] = handler; \
100 SET_HANDLER(_h); \
101 } while (0)
102
103 #define TS_POP_HANDLER handler = handler_stack[--ihandler_stack]
104 #define TS_POP_CALL(_event, _data) handleEvent((TS_POP_HANDLER, _event), _data)
105 #define TS_SET_CALL(_h, _event, _data) handleEvent((SET_HANDLER(_h), _event), _data)
106 #define ASCII_RESPONSE(_s) ascii_response((_s "\r\n"), sizeof(_s "\r\n") - 1)
107 #define ASCII_ERROR() ascii_response(("ERROR\r\n"), sizeof("ERROR\r\n") - 1)
108 #define ASCII_CLIENT_ERROR(_s) ascii_response(("CLIENT_ERROR: " _s "\r\n"), sizeof("CLIENT_ERROR: " _s "\r\n") - 1)
109 #define ASCII_SERVER_ERROR(_s) ascii_response(("SERVER_ERROR: " _s "\r\n"), sizeof("SERVER_ERROR: " _s "\r\n") - 1)
110 #define STRCMP(_s, _const_string) strncmp(_s, _const_string "", sizeof(_const_string) - 1)
111
112 struct MC : Continuation {
113 Action *pending_action;
114 int ihandler_stack;
115 int swallow_bytes;
116 int64_t exptime;
117 ContinuationHandler handler_stack[2];
118 VConnection *nvc;
119 MIOBuffer *rbuf, *wbuf, *cbuf;
120 VIO *rvio, *wvio;
121 IOBufferReader *reader, *writer, *creader;
122 CacheVConnection *crvc, *cwvc;
123 VIO *crvio, *cwvio;
124 CacheKey cache_key;
125 MCCacheHeader *rcache_header, *wcache_header;
126 MCCacheHeader header;
127 char tmp_cache_header_key[256];
128 protocol_binary_request_header binary_header;
129 union {
130 protocol_binary_response_get get;
131 } res;
132 char *key, *tbuf;
133 int read_offset;
134 int end_of_cmd; // -1 means that it is already consumed
135 int ngets;
136 char tmp_cmd_buffer[TSMEMCACHE_TMP_CMD_BUFFER_SIZE];
137 union {
138 struct {
139 unsigned int noreply : 1;
140 unsigned int return_cas : 1;
141 unsigned int set_add : 1;
142 unsigned int set_cas : 1;
143 unsigned int set_append : 1;
144 unsigned int set_prepend : 1;
145 unsigned int set_replace : 1;
146 unsigned int set_incr : 1;
147 unsigned int set_decr : 1;
148 } f;
149 unsigned int ff;
150 };
151 uint64_t nbytes;
152 uint64_t delta;
153
154 static int32_t verbosity;
155 static ink_hrtime last_flush;
156 static int64_t next_cas;
157
158 int write_to_client(int64_t ntowrite = -1);
159 int write_then_read_from_client(int64_t ntowrite = -1);
160 int stream_then_read_from_client(int64_t ntowrite);
161 int write_then_close(int64_t ntowrite = -1);
162 int read_from_client();
163 int get_item();
164 int set_item();
165 int delete_item();
166 int read_from_client_event(int event, void *data);
167 int swallow_then_read_event(int event, void *data);
168 int swallow_cmd_then_read_from_client_event(int event, void *data);
169 int read_binary_from_client_event(int event, void *data);
170 int read_ascii_from_client_event(int event, void *data);
171 int binary_get_event(int event, void *data);
172 int cache_read_event(int event, void *data);
173 int write_then_close_event(int event, void *data);
174 int stream_event(int event, void *data); // cache <=> client
175 int tunnel_event(int event, void *data); // cache <=> cache
176
177 char *get_ascii_input(int n, int *end);
178 int get_ascii_key(char *s, char *e);
179 int ascii_response(const char *s, int len);
180 int ascii_get(char *s, char *e);
181 int ascii_gets();
182 int ascii_set(char *s, char *e);
183 int ascii_delete(char *s, char *e);
184 int ascii_incr_decr(char *s, char *e);
185 int ascii_get_event(int event, void *data);
186 int ascii_set_event(int event, void *data);
187 int ascii_delete_event(int event, void *data);
188 int ascii_incr_decr_event(int event, void *data);
189
190 int write_binary_error(protocol_binary_response_status err, int swallow);
191 void add_binary_header(uint16_t err, uint8_t hdr_len, uint16_t key_len, uint32_t body_len);
192 int write_binary_response(const void *d, int hlen, int keylen, int dlen);
193 int protocol_error();
194 int bin_read_key();
195
196 void new_connection(NetVConnection *netvc, EThread *thread);
197 int unexpected_event();
198 int die();
199 };
200
201 int init_tsmemcache(int port = 11211);
202
203 // INLINE FUNCTIONS
204
205 static inline char *
xutoa(uint32_t i,char * e)206 xutoa(uint32_t i, char *e)
207 {
208 do {
209 *--e = (char)(i % 10 + 48);
210 } while ((i /= 10) > 0);
211 return e;
212 }
213
214 static inline char *
xutoa(uint64_t i,char * e)215 xutoa(uint64_t i, char *e)
216 {
217 do {
218 *--e = (char)(i % 10 + 48);
219 } while ((i /= 10) > 0);
220 return e;
221 }
222
223 static inline uint64_t
xatoull(char * s,char * e)224 xatoull(char *s, char *e)
225 {
226 uint64_t n = 0;
227 if (isdigit(*s)) {
228 n = *s - '0';
229 s++;
230 if (s >= e) {
231 return n;
232 }
233 }
234 while (isdigit(*s)) {
235 n *= 10;
236 n += *s - '0';
237 s++;
238 }
239 return n;
240 }
241