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. */ |
10 | void 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. */ |
43 | int 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. */ |
65 | void 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 */ |
86 | void 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 */ |
129 | void 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; |
d85a09df |
142 | if (timeout <= 0) timeout = 1000; |
571e257d |
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 | } |
26a48504 |
160 | if ((aeWait(fd,AE_WRITABLE,timeout) & AE_WRITABLE) == 0) { |
ae3aeca8 |
161 | close(fd); |
37cc07dd |
162 | addReplySds(c,sdsnew("-IOERR error or timeout connecting to the client\r\n")); |
571e257d |
163 | return; |
164 | } |
165 | |
bff31e12 |
166 | /* Create RESTORE payload and generate the protocol to call the command. */ |
571e257d |
167 | rioInitWithBuffer(&cmd,sdsempty()); |
168 | redisAssertWithInfo(c,NULL,rioWriteBulkCount(&cmd,'*',2)); |
169 | redisAssertWithInfo(c,NULL,rioWriteBulkString(&cmd,"SELECT",6)); |
170 | redisAssertWithInfo(c,NULL,rioWriteBulkLongLong(&cmd,dbid)); |
171 | |
fb8409a5 |
172 | expireat = getExpire(c->db,c->argv[3]); |
173 | if (expireat != -1) { |
174 | ttl = expireat-mstime(); |
175 | if (ttl < 1) ttl = 1; |
176 | } |
571e257d |
177 | redisAssertWithInfo(c,NULL,rioWriteBulkCount(&cmd,'*',4)); |
178 | redisAssertWithInfo(c,NULL,rioWriteBulkString(&cmd,"RESTORE",7)); |
179 | redisAssertWithInfo(c,NULL,c->argv[3]->encoding == REDIS_ENCODING_RAW); |
180 | redisAssertWithInfo(c,NULL,rioWriteBulkString(&cmd,c->argv[3]->ptr,sdslen(c->argv[3]->ptr))); |
3ba5eab7 |
181 | redisAssertWithInfo(c,NULL,rioWriteBulkLongLong(&cmd,ttl)); |
571e257d |
182 | |
183 | /* Finally the last argument that is the serailized object payload |
bff31e12 |
184 | * in the DUMP format. */ |
185 | createDumpPayload(&payload,o); |
186 | redisAssertWithInfo(c,NULL,rioWriteBulkString(&cmd,payload.io.buffer.ptr, |
187 | sdslen(payload.io.buffer.ptr))); |
571e257d |
188 | sdsfree(payload.io.buffer.ptr); |
189 | |
190 | /* Tranfer the query to the other node in 64K chunks. */ |
191 | { |
192 | sds buf = cmd.io.buffer.ptr; |
193 | size_t pos = 0, towrite; |
194 | int nwritten = 0; |
195 | |
196 | while ((towrite = sdslen(buf)-pos) > 0) { |
197 | towrite = (towrite > (64*1024) ? (64*1024) : towrite); |
bde80cb2 |
198 | nwritten = syncWrite(fd,buf+pos,towrite,timeout); |
571e257d |
199 | if (nwritten != (signed)towrite) goto socket_wr_err; |
200 | pos += nwritten; |
201 | } |
202 | } |
203 | |
204 | /* Read back the reply. */ |
205 | { |
206 | char buf1[1024]; |
207 | char buf2[1024]; |
208 | |
209 | /* Read the two replies */ |
210 | if (syncReadLine(fd, buf1, sizeof(buf1), timeout) <= 0) |
211 | goto socket_rd_err; |
212 | if (syncReadLine(fd, buf2, sizeof(buf2), timeout) <= 0) |
213 | goto socket_rd_err; |
214 | if (buf1[0] == '-' || buf2[0] == '-') { |
215 | addReplyErrorFormat(c,"Target instance replied with error: %s", |
216 | (buf1[0] == '-') ? buf1+1 : buf2+1); |
217 | } else { |
218 | robj *aux; |
219 | |
220 | dbDelete(c->db,c->argv[3]); |
221 | signalModifiedKey(c->db,c->argv[3]); |
222 | addReply(c,shared.ok); |
223 | server.dirty++; |
224 | |
225 | /* Translate MIGRATE as DEL for replication/AOF. */ |
226 | aux = createStringObject("DEL",3); |
227 | rewriteClientCommandVector(c,2,aux,c->argv[3]); |
228 | decrRefCount(aux); |
229 | } |
230 | } |
231 | |
232 | sdsfree(cmd.io.buffer.ptr); |
233 | close(fd); |
234 | return; |
235 | |
236 | socket_wr_err: |
37cc07dd |
237 | addReplySds(c,sdsnew("-IOERR error or timeout writing to target instance\r\n")); |
571e257d |
238 | sdsfree(cmd.io.buffer.ptr); |
239 | close(fd); |
240 | return; |
241 | |
242 | socket_rd_err: |
37cc07dd |
243 | addReplySds(c,sdsnew("-IOERR error or timeout reading from target node\r\n")); |
571e257d |
244 | sdsfree(cmd.io.buffer.ptr); |
245 | close(fd); |
246 | return; |
247 | } |