redis server internal cmd_process

Command Process

레디스 서버 교육 신청	레디스 정기점검/기술지원 Redis Technical Support	레디스 엔터프라이즈 서버 Redis Enterprise Server

Command Process

Redis 명령 처리(실행) 과정을 함수로 표시했습니다.
이 문서는 버전 7.0.8을 기준으로 작성했습니다.

명령 처리 흐름

전파(AOF/Repl) 흐름

관련 함수들

createClient() networking.c

connEnableTcpNoDelay(conn); if (server.tcpkeepalive) connKeepAlive(conn,server.tcpkeepalive); connSetReadHandler(conn, readQueryFromClient); connSetPrivateData(conn, c); clientSetDefaultAuth(c);

readQueryFromClient() networking.c

connRead() processInputBuffer()

processInputBuffer() networking.c

while () { if (c->reqtype == PROTO_REQ_INLINE) { if (processInlineBuffer(c) != C_OK) break; } else if (c->reqtype == PROTO_REQ_MULTIBULK) { if (processMultibulkBuffer(c) != C_OK) break; } processCommandAndResetClient() }

processCommandAndResetClient() networking.c

redis-6.0에서 추가 if (processCommand(c) == C_OK) { commandProcessed(c); updateClientMemUsageAndBucket(c); }

processCommand() server.c

c->cmd = c->lastcmd = c->realcmd = lookupCommand(c->argv,c->argc); CMD_READONLY -> is_read_command CMD_WRITE -> is_write_command CMD_DENYOOM -> is_denyoom_command CMD_STALE -> is_denystale_command CMD_LOADING -> is_denyloading_command CMD_WRITE | CMD_MAY_REPLICATE -> is_may_replicate_command CMD_NO_ASYNC_LOADING -> is_deny_async_loading_command authRequired(c) call(c,CMD_CALL_FULL)

call() server.c

c->cmd->proc(c); // 명령 실행 aof에 전파 설정 PROPAGATE_AOF 복제에 전파 설정 PROPAGATE_REPL alsoPropagate() // 전파 가능 여부 확인 afterCommand() -> propagatePendingCommands() -> propagateNow() // AOF/복제에 전파

afterCommand() server.c

propagatePendingCommands()

propagatePendingCommands() server.c

propagateNow()

propagateNow() server.c

feedAppendOnlyFile() // AOF에 전파 replicationFeedSlaves() // 복제에 전파

alsoPropagate() server.c

shouldPropagate() // 전파 여부 확인 redisOpArrayAppend() // 추가로 전파할 명령이 있으면 추가

shouldPropagate() server.c

AOF, slaves 등 전파 여부를 확인한다. 전파할거면 1, 아니면 0을 리턴한다.

redisOpArrayAppend(), redisOpArrayInit(), redisOpArrayFree()

typedef struct redisOp { robj **argv; int argc, dbid, target; } redisOp; typedef struct redisOpArray { redisOp *ops; int numops; int capacity; } redisOpArray;

networking.c Functions

소스 파일 networking.c에 있는 함수들입니다. 버전 7.0.8 기준으로 작성했습니다.

1. sdsZmallocSize()
2. getStringObjectSdsUsedMemory() -> sdsZmallocSize(o->ptr);
3. getStringObjectLen() -> sdslen(o->ptr);
4. dupClientReplyValue()
5. freeClientReplyValue()
6. istMatchObjects()
7. linkClient() -> listAddNodeTail(server.clients,c);
8. clientSetDefaultAuth() -> c->user = DefaultUser;
9. authRequired()
10. createClient() -> connSetReadHandler(conn, readQueryFromClient);
11. installClientWriteHandler() -> connSetWriteHandlerWithBarrier(c->conn, sendReplyToClient, ae_barrier)
12. putClientInPendingWriteQueue() -> listAddNodeHead(server.clients_pending_write,c);
13. prepareClientToWrite() -> putClientInPendingWriteQueue(c);
14. ----- Low level functions to add more data to output buffers. -----
15. _addReplyToBuffer() -> memcpy(c->buf+c->bufpos, s, reply_len);
16. _addReplyProtoToList()
    if (tail) memcpy(tail->buf + tail->used, s, copy);
    if (len) memcpy(tail->buf, s, len);
    
17. _addReplyToBufferOrList()
    size_t reply_len = _addReplyToBuffer(c,s,len);
    if (len > reply_len) _addReplyProtoToList(c,s+reply_len,len-reply_len);
18. addReply()
    prepareClientToWrite()
    _addReplyToBufferOrList()
19. addReplySds()
    prepareClientToWrite()
    _addReplyToBufferOrList()
20. addReplyProto()
    prepareClientToWrite()
    _addReplyToBufferOrList()
21. addReplyErrorLength()
22. afterErrorReply()
23. addReplyErrorObject()
24. addReplyOrErrorObject()
25. addReplyError()
26. addReplyErrorSdsEx()
27. addReplyErrorSds()
28. addReplyErrorSdsSafe()
29. addReplyErrorFormatInternal()
30. addReplyErrorFormatEx()
31. addReplyErrorFormat()
32. addReplyErrorArity()
33. addReplyErrorExpireTime()
34. addReplyStatusLength()
35. addReplyStatus()
36. addReplyStatusFormat()
37. trimReplyUnusedTailSpace()
38. addReplyDeferredLen()
39. setDeferredReply()
40. setDeferredAggregateLen()
41. setDeferredArrayLen()
42. setDeferredMapLen()
43. setDeferredSetLen()
44. setDeferredAttributeLen()
45. setDeferredPushLen()
46. addReplyDouble()
47. addReplyBigNum()
48. addReplyHumanLongDouble()
49. addReplyLongLongWithPrefix()
50. addReplyLongLong()
51. addReplyAggregateLen()
52. addReplyArrayLen()
53. addReplyMapLen()
54. addReplySetLen()
55. addReplyAttributeLen()
56. addReplyPushLen()
57. addReplyNull()
58. addReplyBool()
59. addReplyNullArray()
60. addReplyBulkLen()
61. addReplyBulk()
62. addReplyBulkCBuffer()
63. addReplyBulkSds()
64. setDeferredReplyBulkSds()
65. addReplyBulkCString()
66. addReplyBulkLongLong()
67. addReplyVerbatim()
68. addReplyHelp()
69. addReplySubcommandSyntaxError()
70. AddReplyFromClient()
71. deferredAfterErrorReply()
72. copyReplicaOutputBuffer()
73. clientHasPendingReplies()
74. islocalClient()
75. clientAcceptHandler()
76. acceptCommonHandler()
77. acceptTcpHandler()
78. acceptTLSHandler()
79. acceptUnixHandler()
80. freeClientOriginalArgv()
81. freeClientArgv()
82. disconnectSlaves()
83. anyOtherSlaveWaitRdb()
84. unlinkClient()
85. clearClientConnectionState()
86. freeClient()
87. freeClientAsync()
88. logInvalidUseAndFreeClientAsync()
89. beforeNextClient()
90. freeClientsInAsyncFreeQueue()
91. lookupClientByID()
92. _writevToClient()
93. _writeToClient()
94. writeToClient()
95. sendReplyToClient()
96. handleClientsWithPendingWrites()
97. resetClient()
98. protectClient()
99. unprotectClient()
100. processInlineBuffer() -> argc와 argv를 만든다.
101. setProtocolError()
102. processMultibulkBuffer()
103. commandProcessed()
104. processCommandAndResetClient()
105. processPendingCommandAndInputBuffer() -> processInputBuffer()
106. processInputBuffer()
107. readQueryFromClient()
     connRead()
     processInputBuffer()
108. genClientAddrString()
109. getClientPeerId()
110. getClientSockname()
111. catClientInfoString()
112. getAllClientsInfoString()
113. clientSetName()
114. clientSetNameOrReply()
115. resetCommand()
116. quitCommand()
117. clientCommand()
118. helloCommand()
119. securityWarningCommand()
120. retainOriginalCommandVector()
121. redactClientCommandArgument()
122. rewriteClientCommandVector()
123. replaceClientCommandVector()
124. rewriteClientCommandArgument()
125. getClientOutputBufferMemoryUsage()
126. getClientMemoryUsage()
127. getClientType()
128. getClientTypeByName()
129. getClientTypeName()
130. checkClientOutputBufferLimits()
131. closeClientOnOutputBufferLimitReached()
132. flushSlavesOutputBuffers()
133. updateClientPauseTypeAndEndTime()
134. unblockPostponedClients()
135. pauseClients()
136. unpauseClients()
137. areClientsPaused()
138. checkClientPauseTimeoutAndReturnIfPaused()
139. processEventsWhileBlocked() // RDB/AOF 로드 중 클라이언트 처리
140. ----- Threaded I/O -----
141. getIOPendingCount()
142. setIOPendingCount()
143. IOThreadMain()
144. initThreadedIO()
145. killIOThreads()
146. startThreadedIO()
147. stopThreadedIOIfNeeded()
148. handleClientsWithPendingWritesUsingThreads()
149. postponeClientRead()
150. handleClientsWithPendingReadsUsingThreads()
151. getClientEvictionLimit()
152. evictClients()