Lines Matching refs:slave
44 void putSlaveOnline(client *slave);
217 client *slave = ln->value; in replicationFeedSlaves() local
218 if (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_START) continue; in replicationFeedSlaves()
219 addReply(slave,selectcmd); in replicationFeedSlaves()
257 client *slave = ln->value; in replicationFeedSlaves() local
260 if (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_START) continue; in replicationFeedSlaves()
267 addReplyMultiBulkLen(slave,argc); in replicationFeedSlaves()
272 addReplyBulk(slave,argv[j]); in replicationFeedSlaves()
296 client *slave = ln->value; in replicationFeedSlavesFromMasterStream() local
299 if (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_START) continue; in replicationFeedSlavesFromMasterStream()
300 addReplyString(slave,buf,buflen); in replicationFeedSlavesFromMasterStream()
419 int replicationSetupSlaveForFullResync(client *slave, long long offset) { in replicationSetupSlaveForFullResync() argument
423 slave->psync_initial_offset = offset; in replicationSetupSlaveForFullResync()
424 slave->replstate = SLAVE_STATE_WAIT_BGSAVE_END; in replicationSetupSlaveForFullResync()
432 if (!(slave->flags & CLIENT_PRE_PSYNC)) { in replicationSetupSlaveForFullResync()
435 if (write(slave->fd,buf,buflen) != buflen) { in replicationSetupSlaveForFullResync()
436 freeClientAsync(slave); in replicationSetupSlaveForFullResync()
594 client *slave = ln->value; in startBgsaveForReplication() local
596 if (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_START) { in startBgsaveForReplication()
597 slave->replstate = REPL_STATE_NONE; in startBgsaveForReplication()
598 slave->flags &= ~CLIENT_SLAVE; in startBgsaveForReplication()
600 addReplyError(slave, in startBgsaveForReplication()
602 slave->flags |= CLIENT_CLOSE_AFTER_REPLY; in startBgsaveForReplication()
613 client *slave = ln->value; in startBgsaveForReplication() local
615 if (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_START) { in startBgsaveForReplication()
616 replicationSetupSlaveForFullResync(slave, in startBgsaveForReplication()
710 client *slave; in syncCommand() local
716 slave = ln->value; in syncCommand()
717 if (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_END) break; in syncCommand()
721 if (ln && ((c->slave_capa & slave->slave_capa) == slave->slave_capa)) { in syncCommand()
724 copyClientOutputBuffer(c,slave); in syncCommand()
725 replicationSetupSlaveForFullResync(c,slave->psync_initial_offset); in syncCommand()
857 void putSlaveOnline(client *slave) { in putSlaveOnline() argument
858 slave->replstate = SLAVE_STATE_ONLINE; in putSlaveOnline()
859 slave->repl_put_online_on_ack = 0; in putSlaveOnline()
860 slave->repl_ack_time = server.unixtime; /* Prevent false timeout. */ in putSlaveOnline()
861 if (aeCreateFileEvent(server.el, slave->fd, AE_WRITABLE, in putSlaveOnline()
862 sendReplyToClient, slave) == AE_ERR) { in putSlaveOnline()
864 freeClient(slave); in putSlaveOnline()
869 replicationGetSlaveName(slave)); in putSlaveOnline()
873 client *slave = privdata; in sendBulkToSlave() local
882 if (slave->replpreamble) { in sendBulkToSlave()
883 nwritten = write(fd,slave->replpreamble,sdslen(slave->replpreamble)); in sendBulkToSlave()
887 freeClient(slave); in sendBulkToSlave()
891 sdsrange(slave->replpreamble,nwritten,-1); in sendBulkToSlave()
892 if (sdslen(slave->replpreamble) == 0) { in sendBulkToSlave()
893 sdsfree(slave->replpreamble); in sendBulkToSlave()
894 slave->replpreamble = NULL; in sendBulkToSlave()
902 lseek(slave->repldbfd,slave->repldboff,SEEK_SET); in sendBulkToSlave()
903 buflen = read(slave->repldbfd,buf,PROTO_IOBUF_LEN); in sendBulkToSlave()
907 freeClient(slave); in sendBulkToSlave()
914 freeClient(slave); in sendBulkToSlave()
918 slave->repldboff += nwritten; in sendBulkToSlave()
920 if (slave->repldboff == slave->repldbsize) { in sendBulkToSlave()
921 close(slave->repldbfd); in sendBulkToSlave()
922 slave->repldbfd = -1; in sendBulkToSlave()
923 aeDeleteFileEvent(server.el,slave->fd,AE_WRITABLE); in sendBulkToSlave()
924 putSlaveOnline(slave); in sendBulkToSlave()
950 client *slave = ln->value; in updateSlavesWaitingBgsave() local
952 if (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_START) { in updateSlavesWaitingBgsave()
954 mincapa = (mincapa == -1) ? slave->slave_capa : in updateSlavesWaitingBgsave()
955 (mincapa & slave->slave_capa); in updateSlavesWaitingBgsave()
956 } else if (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_END) { in updateSlavesWaitingBgsave()
967 replicationGetSlaveName(slave)); in updateSlavesWaitingBgsave()
973 slave->replstate = SLAVE_STATE_ONLINE; in updateSlavesWaitingBgsave()
974 slave->repl_put_online_on_ack = 1; in updateSlavesWaitingBgsave()
975 slave->repl_ack_time = server.unixtime; /* Timeout otherwise. */ in updateSlavesWaitingBgsave()
978 freeClient(slave); in updateSlavesWaitingBgsave()
982 if ((slave->repldbfd = open(server.rdb_filename,O_RDONLY)) == -1 || in updateSlavesWaitingBgsave()
983 redis_fstat(slave->repldbfd,&buf) == -1) { in updateSlavesWaitingBgsave()
984 freeClient(slave); in updateSlavesWaitingBgsave()
988 slave->repldboff = 0; in updateSlavesWaitingBgsave()
989 slave->repldbsize = buf.st_size; in updateSlavesWaitingBgsave()
990 slave->replstate = SLAVE_STATE_SEND_BULK; in updateSlavesWaitingBgsave()
991 slave->replpreamble = sdscatprintf(sdsempty(),"$%lld\r\n", in updateSlavesWaitingBgsave()
992 (unsigned long long) slave->repldbsize); in updateSlavesWaitingBgsave()
994 aeDeleteFileEvent(server.el,slave->fd,AE_WRITABLE); in updateSlavesWaitingBgsave()
995 … if (aeCreateFileEvent(server.el, slave->fd, AE_WRITABLE, sendBulkToSlave, slave) == AE_ERR) { in updateSlavesWaitingBgsave()
996 freeClient(slave); in updateSlavesWaitingBgsave()
2091 client *slave = ln->value; in roleCommand() local
2092 char ip[NET_IP_STR_LEN], *slaveip = slave->slave_ip; in roleCommand()
2095 if (anetPeerToString(slave->fd,ip,sizeof(ip),NULL) == -1) in roleCommand()
2099 if (slave->replstate != SLAVE_STATE_ONLINE) continue; in roleCommand()
2102 addReplyBulkLongLong(c,slave->slave_listening_port); in roleCommand()
2103 addReplyBulkLongLong(c,slave->repl_ack_off); in roleCommand()
2290 client *slave = ln->value; in refreshGoodSlavesCount() local
2291 time_t lag = server.unixtime - slave->repl_ack_time; in refreshGoodSlavesCount()
2293 if (slave->replstate == SLAVE_STATE_ONLINE && in refreshGoodSlavesCount()
2426 client *slave = ln->value; in replicationCountAcksByOffset() local
2428 if (slave->replstate != SLAVE_STATE_ONLINE) continue; in replicationCountAcksByOffset()
2429 if (slave->repl_ack_off >= offset) count++; in replicationCountAcksByOffset()
2630 client *slave = ln->value; in replicationCron() local
2633 (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_START || in replicationCron()
2634 (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_END && in replicationCron()
2638 if (write(slave->fd, "\n", 1) == -1) { in replicationCron()
2651 client *slave = ln->value; in replicationCron() local
2653 if (slave->replstate != SLAVE_STATE_ONLINE) continue; in replicationCron()
2654 if (slave->flags & CLIENT_PRE_PSYNC) continue; in replicationCron()
2655 if ((server.unixtime - slave->repl_ack_time) > server.repl_timeout) in replicationCron()
2658 replicationGetSlaveName(slave)); in replicationCron()
2659 freeClient(slave); in replicationCron()
2726 client *slave = ln->value; in replicationCron() local
2727 if (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_START) { in replicationCron()
2728 idle = server.unixtime - slave->lastinteraction; in replicationCron()
2731 mincapa = (mincapa == -1) ? slave->slave_capa : in replicationCron()
2732 (mincapa & slave->slave_capa); in replicationCron()