[redis.git] / src / migrate.c

#include "redis.h"
#include "endianconv.h"

/* -----------------------------------------------------------------------------
 * DUMP, RESTORE and MIGRATE commands
 * -------------------------------------------------------------------------- */

/* Generates a DUMP-format representation of the object 'o', adding it to the
 * io stream pointed by 'rio'. This function can't fail. */
void createDumpPayload(rio *payload, robj *o) {
    unsigned char buf[2];
    uint64_t crc;

    /* Serialize the object in a RDB-like format. It consist of an object type
     * byte followed by the serialized object. This is understood by RESTORE. */
    rioInitWithBuffer(payload,sdsempty());
    redisAssert(rdbSaveObjectType(payload,o));
    redisAssert(rdbSaveObject(payload,o));

    /* Write the footer, this is how it looks like:
     * ----------------+---------------------+---------------+
     * ... RDB payload | 2 bytes RDB version | 8 bytes CRC64 |
     * ----------------+---------------------+---------------+
     * RDB version and CRC are both in little endian.
     */

    /* RDB version */
    buf[0] = REDIS_RDB_VERSION & 0xff;
    buf[1] = (REDIS_RDB_VERSION >> 8) & 0xff;
    payload->io.buffer.ptr = sdscatlen(payload->io.buffer.ptr,buf,2);

    /* CRC64 */
    crc = crc64(0,(unsigned char*)payload->io.buffer.ptr,
                sdslen(payload->io.buffer.ptr));
    memrev64ifbe(&crc);
    payload->io.buffer.ptr = sdscatlen(payload->io.buffer.ptr,&crc,8);
}

/* Verify that the RDB version of the dump payload matches the one of this Redis
 * instance and that the checksum is ok.
 * If the DUMP payload looks valid REDIS_OK is returned, otherwise REDIS_ERR
 * is returned. */
int verifyDumpPayload(unsigned char *p, size_t len) {
    unsigned char *footer;
    uint16_t rdbver;
    uint64_t crc;

    /* At least 2 bytes of RDB version and 8 of CRC64 should be present. */
    if (len < 10) return REDIS_ERR;
    footer = p+(len-10);

    /* Verify RDB version */
    rdbver = (footer[1] << 8) | footer[0];
    if (rdbver != REDIS_RDB_VERSION) return REDIS_ERR;

    /* Verify CRC64 */
    crc = crc64(0,p,len-8);
    memrev64ifbe(&crc);
    return (memcmp(&crc,footer+2,8) == 0) ? REDIS_OK : REDIS_ERR;
}

/* DUMP keyname
 * DUMP is actually not used by Redis Cluster but it is the obvious
 * complement of RESTORE and can be useful for different applications. */
void dumpCommand(redisClient *c) {
    robj *o, *dumpobj;
    rio payload;

    /* Check if the key is here. */
    if ((o = lookupKeyRead(c->db,c->argv[1])) == NULL) {
        addReply(c,shared.nullbulk);
        return;
    }

    /* Create the DUMP encoded representation. */
    createDumpPayload(&payload,o);

    /* Transfer to the client */
    dumpobj = createObject(REDIS_STRING,payload.io.buffer.ptr);
    addReplyBulk(c,dumpobj);
    decrRefCount(dumpobj);
    return;
}

/* RESTORE key ttl serialized-value */
void restoreCommand(redisClient *c) {
    long ttl;
    rio payload;
    int type;
    robj *obj;

    /* Make sure this key does not already exist here... */
    if (lookupKeyWrite(c->db,c->argv[1]) != NULL) {
        addReplyError(c,"Target key name is busy.");
        return;
    }

    /* Check if the TTL value makes sense */
    if (getLongFromObjectOrReply(c,c->argv[2],&ttl,NULL) != REDIS_OK) {
        return;
    } else if (ttl < 0) {
        addReplyError(c,"Invalid TTL value, must be >= 0");
        return;
    }

    /* Verify RDB version and data checksum. */
    if (verifyDumpPayload(c->argv[3]->ptr,sdslen(c->argv[3]->ptr)) == REDIS_ERR) {
        addReplyError(c,"DUMP payload version or checksum are wrong");
        return;
    }

    rioInitWithBuffer(&payload,c->argv[3]->ptr);
    if (((type = rdbLoadObjectType(&payload)) == -1) ||
        ((obj = rdbLoadObject(type,&payload)) == NULL))
    {
        addReplyError(c,"Bad data format");
        return;
    }

    /* Create the key and set the TTL if any */
    dbAdd(c->db,c->argv[1],obj);
    if (ttl) setExpire(c->db,c->argv[1],mstime()+ttl);
    signalModifiedKey(c->db,c->argv[1]);
    addReply(c,shared.ok);
    server.dirty++;
}

/* MIGRATE host port key dbid timeout */
void migrateCommand(redisClient *c) {
    int fd;
    long timeout;
    long dbid;
    long long ttl, expireat;
    robj *o;
    rio cmd, payload;

    /* Sanity check */
    if (getLongFromObjectOrReply(c,c->argv[5],&timeout,NULL) != REDIS_OK)
        return;
    if (getLongFromObjectOrReply(c,c->argv[4],&dbid,NULL) != REDIS_OK)
        return;
    if (timeout <= 0) timeout = 1;

    /* Check if the key is here. If not we reply with success as there is
     * nothing to migrate (for instance the key expired in the meantime), but
     * we include such information in the reply string. */
    if ((o = lookupKeyRead(c->db,c->argv[3])) == NULL) {
        addReplySds(c,sdsnew("+NOKEY\r\n"));
        return;
    }
    
    /* Connect */
    fd = anetTcpNonBlockConnect(server.neterr,c->argv[1]->ptr,
                atoi(c->argv[2]->ptr));
    if (fd == -1) {
        addReplyErrorFormat(c,"Can't connect to target node: %s",
            server.neterr);
        return;
    }
    if ((aeWait(fd,AE_WRITABLE,timeout*1000) & AE_WRITABLE) == 0) {
        addReplySds(c,sdsnew("-IOERR error or timeout connecting to the client\r\n"));
        return;
    }

    /* Create RESTORE payload and generate the protocol to call the command. */
    rioInitWithBuffer(&cmd,sdsempty());
    redisAssertWithInfo(c,NULL,rioWriteBulkCount(&cmd,'*',2));
    redisAssertWithInfo(c,NULL,rioWriteBulkString(&cmd,"SELECT",6));
    redisAssertWithInfo(c,NULL,rioWriteBulkLongLong(&cmd,dbid));

    expireat = getExpire(c->db,c->argv[3]);
    if (expireat != -1) {
        ttl = expireat-mstime();
        if (ttl < 1) ttl = 1;
    }
    redisAssertWithInfo(c,NULL,rioWriteBulkCount(&cmd,'*',4));
    redisAssertWithInfo(c,NULL,rioWriteBulkString(&cmd,"RESTORE",7));
    redisAssertWithInfo(c,NULL,c->argv[3]->encoding == REDIS_ENCODING_RAW);
    redisAssertWithInfo(c,NULL,rioWriteBulkString(&cmd,c->argv[3]->ptr,sdslen(c->argv[3]->ptr)));
    redisAssertWithInfo(c,NULL,rioWriteBulkLongLong(&cmd,(expireat==-1) ? 0 : ttl));

    /* Finally the last argument that is the serailized object payload
     * in the DUMP format. */
    createDumpPayload(&payload,o);
    redisAssertWithInfo(c,NULL,rioWriteBulkString(&cmd,payload.io.buffer.ptr,
                                sdslen(payload.io.buffer.ptr)));
    sdsfree(payload.io.buffer.ptr);

    /* Tranfer the query to the other node in 64K chunks. */
    {
        sds buf = cmd.io.buffer.ptr;
        size_t pos = 0, towrite;
        int nwritten = 0;

        while ((towrite = sdslen(buf)-pos) > 0) {
            towrite = (towrite > (64*1024) ? (64*1024) : towrite);
            nwritten = syncWrite(fd,buf+pos,towrite,timeout);
            if (nwritten != (signed)towrite) goto socket_wr_err;
            pos += nwritten;
        }
    }

    /* Read back the reply. */
    {
        char buf1[1024];
        char buf2[1024];

        /* Read the two replies */
        if (syncReadLine(fd, buf1, sizeof(buf1), timeout) <= 0)
            goto socket_rd_err;
        if (syncReadLine(fd, buf2, sizeof(buf2), timeout) <= 0)
            goto socket_rd_err;
        if (buf1[0] == '-' || buf2[0] == '-') {
            addReplyErrorFormat(c,"Target instance replied with error: %s",
                (buf1[0] == '-') ? buf1+1 : buf2+1);
        } else {
            robj *aux;

            dbDelete(c->db,c->argv[3]);
            signalModifiedKey(c->db,c->argv[3]);
            addReply(c,shared.ok);
            server.dirty++;

            /* Translate MIGRATE as DEL for replication/AOF. */
            aux = createStringObject("DEL",3);
            rewriteClientCommandVector(c,2,aux,c->argv[3]);
            decrRefCount(aux);
        }
    }

    sdsfree(cmd.io.buffer.ptr);
    close(fd);
    return;

socket_wr_err:
    addReplySds(c,sdsnew("-IOERR error or timeout writing to target instance\r\n"));
    sdsfree(cmd.io.buffer.ptr);
    close(fd);
    return;

socket_rd_err:
    addReplySds(c,sdsnew("-IOERR error or timeout reading from target node\r\n"));
    sdsfree(cmd.io.buffer.ptr);
    close(fd);
    return;
}
Commit	Line	Data
571e257d	1	#include "redis.h"
bff31e12	2	#include "endianconv.h"
571e257d	3
571e257d	4	/* -----------------------------------------------------------------------------
bff31e12	5	* DUMP, RESTORE and MIGRATE commands
571e257d	6	* -------------------------------------------------------------------------- */
571e257d	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;
fb8409a5	133	long long ttl, 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)));
fb8409a5	180	redisAssertWithInfo(c,NULL,rioWriteBulkLongLong(&cmd,(expireat==-1) ? 0 : ttl));
571e257d	181
571e257d	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 > (641024) ? (641024) : 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
	235	socket_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
	241	socket_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	}