]> git.saurik.com Git - redis.git/blame - src/migrate.c
popcount() optimization for speed.
[redis.git] / src / migrate.c
CommitLineData
571e257d 1#include "redis.h"
bff31e12 2#include "endianconv.h"
571e257d 3
4/* -----------------------------------------------------------------------------
bff31e12 5 * DUMP, RESTORE and MIGRATE commands
571e257d 6 * -------------------------------------------------------------------------- */
7
bff31e12 8/* Generates a DUMP-format representation of the object 'o', adding it to the
9 * io stream pointed by 'rio'. This function can't fail. */
10void createDumpPayload(rio *payload, robj *o) {
11 unsigned char buf[2];
12 uint64_t crc;
13
14 /* Serialize the object in a RDB-like format. It consist of an object type
15 * byte followed by the serialized object. This is understood by RESTORE. */
16 rioInitWithBuffer(payload,sdsempty());
17 redisAssert(rdbSaveObjectType(payload,o));
18 redisAssert(rdbSaveObject(payload,o));
19
20 /* Write the footer, this is how it looks like:
21 * ----------------+---------------------+---------------+
22 * ... RDB payload | 2 bytes RDB version | 8 bytes CRC64 |
23 * ----------------+---------------------+---------------+
24 * RDB version and CRC are both in little endian.
25 */
26
27 /* RDB version */
28 buf[0] = REDIS_RDB_VERSION & 0xff;
29 buf[1] = (REDIS_RDB_VERSION >> 8) & 0xff;
30 payload->io.buffer.ptr = sdscatlen(payload->io.buffer.ptr,buf,2);
31
32 /* CRC64 */
e9574039 33 crc = crc64(0,(unsigned char*)payload->io.buffer.ptr,
bff31e12 34 sdslen(payload->io.buffer.ptr));
35 memrev64ifbe(&crc);
36 payload->io.buffer.ptr = sdscatlen(payload->io.buffer.ptr,&crc,8);
37}
38
39/* Verify that the RDB version of the dump payload matches the one of this Redis
40 * instance and that the checksum is ok.
41 * If the DUMP payload looks valid REDIS_OK is returned, otherwise REDIS_ERR
42 * is returned. */
43int verifyDumpPayload(unsigned char *p, size_t len) {
44 unsigned char *footer;
45 uint16_t rdbver;
46 uint64_t crc;
47
48 /* At least 2 bytes of RDB version and 8 of CRC64 should be present. */
49 if (len < 10) return REDIS_ERR;
50 footer = p+(len-10);
51
52 /* Verify RDB version */
53 rdbver = (footer[1] << 8) | footer[0];
54 if (rdbver != REDIS_RDB_VERSION) return REDIS_ERR;
55
56 /* Verify CRC64 */
e9574039 57 crc = crc64(0,p,len-8);
bff31e12 58 memrev64ifbe(&crc);
59 return (memcmp(&crc,footer+2,8) == 0) ? REDIS_OK : REDIS_ERR;
60}
61
62/* DUMP keyname
63 * DUMP is actually not used by Redis Cluster but it is the obvious
64 * complement of RESTORE and can be useful for different applications. */
65void dumpCommand(redisClient *c) {
66 robj *o, *dumpobj;
67 rio payload;
68
69 /* Check if the key is here. */
70 if ((o = lookupKeyRead(c->db,c->argv[1])) == NULL) {
71 addReply(c,shared.nullbulk);
72 return;
73 }
74
75 /* Create the DUMP encoded representation. */
76 createDumpPayload(&payload,o);
77
78 /* Transfer to the client */
79 dumpobj = createObject(REDIS_STRING,payload.io.buffer.ptr);
80 addReplyBulk(c,dumpobj);
81 decrRefCount(dumpobj);
82 return;
83}
84
571e257d 85/* RESTORE key ttl serialized-value */
86void restoreCommand(redisClient *c) {
87 long ttl;
88 rio payload;
89 int type;
90 robj *obj;
91
92 /* Make sure this key does not already exist here... */
93 if (lookupKeyWrite(c->db,c->argv[1]) != NULL) {
94 addReplyError(c,"Target key name is busy.");
95 return;
96 }
97
98 /* Check if the TTL value makes sense */
99 if (getLongFromObjectOrReply(c,c->argv[2],&ttl,NULL) != REDIS_OK) {
100 return;
101 } else if (ttl < 0) {
102 addReplyError(c,"Invalid TTL value, must be >= 0");
103 return;
104 }
105
bff31e12 106 /* Verify RDB version and data checksum. */
107 if (verifyDumpPayload(c->argv[3]->ptr,sdslen(c->argv[3]->ptr)) == REDIS_ERR) {
108 addReplyError(c,"DUMP payload version or checksum are wrong");
109 return;
110 }
111
571e257d 112 rioInitWithBuffer(&payload,c->argv[3]->ptr);
113 if (((type = rdbLoadObjectType(&payload)) == -1) ||
114 ((obj = rdbLoadObject(type,&payload)) == NULL))
115 {
116 addReplyError(c,"Bad data format");
117 return;
118 }
119
120 /* Create the key and set the TTL if any */
121 dbAdd(c->db,c->argv[1],obj);
bff31e12 122 if (ttl) setExpire(c->db,c->argv[1],mstime()+ttl);
571e257d 123 signalModifiedKey(c->db,c->argv[1]);
124 addReply(c,shared.ok);
125 server.dirty++;
126}
127
128/* MIGRATE host port key dbid timeout */
129void migrateCommand(redisClient *c) {
130 int fd;
131 long timeout;
132 long dbid;
3ba5eab7 133 long long ttl = 0, expireat;
571e257d 134 robj *o;
135 rio cmd, payload;
136
137 /* Sanity check */
138 if (getLongFromObjectOrReply(c,c->argv[5],&timeout,NULL) != REDIS_OK)
139 return;
140 if (getLongFromObjectOrReply(c,c->argv[4],&dbid,NULL) != REDIS_OK)
141 return;
142 if (timeout <= 0) timeout = 1;
143
144 /* Check if the key is here. If not we reply with success as there is
145 * nothing to migrate (for instance the key expired in the meantime), but
146 * we include such information in the reply string. */
147 if ((o = lookupKeyRead(c->db,c->argv[3])) == NULL) {
148 addReplySds(c,sdsnew("+NOKEY\r\n"));
149 return;
150 }
151
152 /* Connect */
153 fd = anetTcpNonBlockConnect(server.neterr,c->argv[1]->ptr,
154 atoi(c->argv[2]->ptr));
155 if (fd == -1) {
156 addReplyErrorFormat(c,"Can't connect to target node: %s",
157 server.neterr);
158 return;
159 }
160 if ((aeWait(fd,AE_WRITABLE,timeout*1000) & AE_WRITABLE) == 0) {
37cc07dd 161 addReplySds(c,sdsnew("-IOERR error or timeout connecting to the client\r\n"));
571e257d 162 return;
163 }
164
bff31e12 165 /* Create RESTORE payload and generate the protocol to call the command. */
571e257d 166 rioInitWithBuffer(&cmd,sdsempty());
167 redisAssertWithInfo(c,NULL,rioWriteBulkCount(&cmd,'*',2));
168 redisAssertWithInfo(c,NULL,rioWriteBulkString(&cmd,"SELECT",6));
169 redisAssertWithInfo(c,NULL,rioWriteBulkLongLong(&cmd,dbid));
170
fb8409a5 171 expireat = getExpire(c->db,c->argv[3]);
172 if (expireat != -1) {
173 ttl = expireat-mstime();
174 if (ttl < 1) ttl = 1;
175 }
571e257d 176 redisAssertWithInfo(c,NULL,rioWriteBulkCount(&cmd,'*',4));
177 redisAssertWithInfo(c,NULL,rioWriteBulkString(&cmd,"RESTORE",7));
178 redisAssertWithInfo(c,NULL,c->argv[3]->encoding == REDIS_ENCODING_RAW);
179 redisAssertWithInfo(c,NULL,rioWriteBulkString(&cmd,c->argv[3]->ptr,sdslen(c->argv[3]->ptr)));
3ba5eab7 180 redisAssertWithInfo(c,NULL,rioWriteBulkLongLong(&cmd,ttl));
571e257d 181
182 /* Finally the last argument that is the serailized object payload
bff31e12 183 * in the DUMP format. */
184 createDumpPayload(&payload,o);
185 redisAssertWithInfo(c,NULL,rioWriteBulkString(&cmd,payload.io.buffer.ptr,
186 sdslen(payload.io.buffer.ptr)));
571e257d 187 sdsfree(payload.io.buffer.ptr);
188
189 /* Tranfer the query to the other node in 64K chunks. */
190 {
191 sds buf = cmd.io.buffer.ptr;
192 size_t pos = 0, towrite;
193 int nwritten = 0;
194
195 while ((towrite = sdslen(buf)-pos) > 0) {
196 towrite = (towrite > (64*1024) ? (64*1024) : towrite);
bde80cb2 197 nwritten = syncWrite(fd,buf+pos,towrite,timeout);
571e257d 198 if (nwritten != (signed)towrite) goto socket_wr_err;
199 pos += nwritten;
200 }
201 }
202
203 /* Read back the reply. */
204 {
205 char buf1[1024];
206 char buf2[1024];
207
208 /* Read the two replies */
209 if (syncReadLine(fd, buf1, sizeof(buf1), timeout) <= 0)
210 goto socket_rd_err;
211 if (syncReadLine(fd, buf2, sizeof(buf2), timeout) <= 0)
212 goto socket_rd_err;
213 if (buf1[0] == '-' || buf2[0] == '-') {
214 addReplyErrorFormat(c,"Target instance replied with error: %s",
215 (buf1[0] == '-') ? buf1+1 : buf2+1);
216 } else {
217 robj *aux;
218
219 dbDelete(c->db,c->argv[3]);
220 signalModifiedKey(c->db,c->argv[3]);
221 addReply(c,shared.ok);
222 server.dirty++;
223
224 /* Translate MIGRATE as DEL for replication/AOF. */
225 aux = createStringObject("DEL",3);
226 rewriteClientCommandVector(c,2,aux,c->argv[3]);
227 decrRefCount(aux);
228 }
229 }
230
231 sdsfree(cmd.io.buffer.ptr);
232 close(fd);
233 return;
234
235socket_wr_err:
37cc07dd 236 addReplySds(c,sdsnew("-IOERR error or timeout writing to target instance\r\n"));
571e257d 237 sdsfree(cmd.io.buffer.ptr);
238 close(fd);
239 return;
240
241socket_rd_err:
37cc07dd 242 addReplySds(c,sdsnew("-IOERR error or timeout reading from target node\r\n"));
571e257d 243 sdsfree(cmd.io.buffer.ptr);
244 close(fd);
245 return;
246}