Lines Matching refs:slave
43 void putSlaveOnline(client *slave);
209 client *slave = ln->value; in replicationFeedSlaves() local
210 if (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_START) continue; in replicationFeedSlaves()
211 addReply(slave,selectcmd); in replicationFeedSlaves()
249 client *slave = ln->value; in replicationFeedSlaves() local
252 if (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_START) continue; in replicationFeedSlaves()
259 addReplyMultiBulkLen(slave,argc); in replicationFeedSlaves()
264 addReplyBulk(slave,argv[j]); in replicationFeedSlaves()
387 int replicationSetupSlaveForFullResync(client *slave, long long offset) { in replicationSetupSlaveForFullResync() argument
391 slave->psync_initial_offset = offset; in replicationSetupSlaveForFullResync()
392 slave->replstate = SLAVE_STATE_WAIT_BGSAVE_END; in replicationSetupSlaveForFullResync()
400 if (!(slave->flags & CLIENT_PRE_PSYNC)) { in replicationSetupSlaveForFullResync()
403 if (write(slave->fd,buf,buflen) != buflen) { in replicationSetupSlaveForFullResync()
404 freeClientAsync(slave); in replicationSetupSlaveForFullResync()
530 client *slave = ln->value; in startBgsaveForReplication() local
532 if (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_START) { in startBgsaveForReplication()
533 slave->flags &= ~CLIENT_SLAVE; in startBgsaveForReplication()
535 addReplyError(slave, in startBgsaveForReplication()
537 slave->flags |= CLIENT_CLOSE_AFTER_REPLY; in startBgsaveForReplication()
548 client *slave = ln->value; in startBgsaveForReplication() local
550 if (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_START) { in startBgsaveForReplication()
551 replicationSetupSlaveForFullResync(slave, in startBgsaveForReplication()
635 client *slave; in syncCommand() local
641 slave = ln->value; in syncCommand()
642 if (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_END) break; in syncCommand()
646 if (ln && ((c->slave_capa & slave->slave_capa) == slave->slave_capa)) { in syncCommand()
649 copyClientOutputBuffer(c,slave); in syncCommand()
650 replicationSetupSlaveForFullResync(c,slave->psync_initial_offset); in syncCommand()
783 void putSlaveOnline(client *slave) { in putSlaveOnline() argument
784 slave->replstate = SLAVE_STATE_ONLINE; in putSlaveOnline()
785 slave->repl_put_online_on_ack = 0; in putSlaveOnline()
786 slave->repl_ack_time = server.unixtime; /* Prevent false timeout. */ in putSlaveOnline()
787 if (aeCreateFileEvent(server.el, slave->fd, AE_WRITABLE, in putSlaveOnline()
788 sendReplyToClient, slave) == AE_ERR) { in putSlaveOnline()
790 freeClient(slave); in putSlaveOnline()
795 replicationGetSlaveName(slave)); in putSlaveOnline()
799 client *slave = privdata; in sendBulkToSlave() local
808 if (slave->replpreamble) { in sendBulkToSlave()
809 nwritten = write(fd,slave->replpreamble,sdslen(slave->replpreamble)); in sendBulkToSlave()
813 freeClient(slave); in sendBulkToSlave()
817 sdsrange(slave->replpreamble,nwritten,-1); in sendBulkToSlave()
818 if (sdslen(slave->replpreamble) == 0) { in sendBulkToSlave()
819 sdsfree(slave->replpreamble); in sendBulkToSlave()
820 slave->replpreamble = NULL; in sendBulkToSlave()
828 lseek(slave->repldbfd,slave->repldboff,SEEK_SET); in sendBulkToSlave()
829 buflen = read(slave->repldbfd,buf,PROTO_IOBUF_LEN); in sendBulkToSlave()
833 freeClient(slave); in sendBulkToSlave()
840 freeClient(slave); in sendBulkToSlave()
844 slave->repldboff += nwritten; in sendBulkToSlave()
846 if (slave->repldboff == slave->repldbsize) { in sendBulkToSlave()
847 close(slave->repldbfd); in sendBulkToSlave()
848 slave->repldbfd = -1; in sendBulkToSlave()
849 aeDeleteFileEvent(server.el,slave->fd,AE_WRITABLE); in sendBulkToSlave()
850 putSlaveOnline(slave); in sendBulkToSlave()
876 client *slave = ln->value; in updateSlavesWaitingBgsave() local
878 if (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_START) { in updateSlavesWaitingBgsave()
880 mincapa = (mincapa == -1) ? slave->slave_capa : in updateSlavesWaitingBgsave()
881 (mincapa & slave->slave_capa); in updateSlavesWaitingBgsave()
882 } else if (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_END) { in updateSlavesWaitingBgsave()
893 replicationGetSlaveName(slave)); in updateSlavesWaitingBgsave()
899 slave->replstate = SLAVE_STATE_ONLINE; in updateSlavesWaitingBgsave()
900 slave->repl_put_online_on_ack = 1; in updateSlavesWaitingBgsave()
901 slave->repl_ack_time = server.unixtime; /* Timeout otherwise. */ in updateSlavesWaitingBgsave()
904 freeClient(slave); in updateSlavesWaitingBgsave()
908 if ((slave->repldbfd = open(server.rdb_filename,O_RDONLY)) == -1 || in updateSlavesWaitingBgsave()
909 redis_fstat(slave->repldbfd,&buf) == -1) { in updateSlavesWaitingBgsave()
910 freeClient(slave); in updateSlavesWaitingBgsave()
914 slave->repldboff = 0; in updateSlavesWaitingBgsave()
915 slave->repldbsize = buf.st_size; in updateSlavesWaitingBgsave()
916 slave->replstate = SLAVE_STATE_SEND_BULK; in updateSlavesWaitingBgsave()
917 slave->replpreamble = sdscatprintf(sdsempty(),"$%lld\r\n", in updateSlavesWaitingBgsave()
918 (unsigned long long) slave->repldbsize); in updateSlavesWaitingBgsave()
920 aeDeleteFileEvent(server.el,slave->fd,AE_WRITABLE); in updateSlavesWaitingBgsave()
921 … if (aeCreateFileEvent(server.el, slave->fd, AE_WRITABLE, sendBulkToSlave, slave) == AE_ERR) { in updateSlavesWaitingBgsave()
922 freeClient(slave); in updateSlavesWaitingBgsave()
1832 client *slave = ln->value; in roleCommand() local
1833 char ip[NET_IP_STR_LEN], *slaveip = slave->slave_ip; in roleCommand()
1836 if (anetPeerToString(slave->fd,ip,sizeof(ip),NULL) == -1) in roleCommand()
1840 if (slave->replstate != SLAVE_STATE_ONLINE) continue; in roleCommand()
1843 addReplyBulkLongLong(c,slave->slave_listening_port); in roleCommand()
1844 addReplyBulkLongLong(c,slave->repl_ack_off); in roleCommand()
1991 client *slave = ln->value; in refreshGoodSlavesCount() local
1992 time_t lag = server.unixtime - slave->repl_ack_time; in refreshGoodSlavesCount()
1994 if (slave->replstate == SLAVE_STATE_ONLINE && in refreshGoodSlavesCount()
2127 client *slave = ln->value; in replicationCountAcksByOffset() local
2129 if (slave->replstate != SLAVE_STATE_ONLINE) continue; in replicationCountAcksByOffset()
2130 if (slave->repl_ack_off >= offset) count++; in replicationCountAcksByOffset()
2305 client *slave = ln->value; in replicationCron() local
2307 if (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_START || in replicationCron()
2308 (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_END && in replicationCron()
2311 if (write(slave->fd, "\n", 1) == -1) { in replicationCron()
2324 client *slave = ln->value; in replicationCron() local
2326 if (slave->replstate != SLAVE_STATE_ONLINE) continue; in replicationCron()
2327 if (slave->flags & CLIENT_PRE_PSYNC) continue; in replicationCron()
2328 if ((server.unixtime - slave->repl_ack_time) > server.repl_timeout) in replicationCron()
2331 replicationGetSlaveName(slave)); in replicationCron()
2332 freeClient(slave); in replicationCron()
2378 client *slave = ln->value; in replicationCron() local
2379 if (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_START) { in replicationCron()
2380 idle = server.unixtime - slave->lastinteraction; in replicationCron()
2383 mincapa = (mincapa == -1) ? slave->slave_capa : in replicationCron()
2384 (mincapa & slave->slave_capa); in replicationCron()