async rpc in put message to cluster #24
Conversation
| @@ -222,9 +224,10 @@ func (self *NsqdCoordinator) PutMessagesToCluster(topic *nsqd.Topic, | |||
| } | |||
| return false | |||
| } | |||
| clusterErr := self.doSyncOpToCluster(true, coord, doLocalWrite, doLocalExit, doLocalCommit, doLocalRollback, | |||
| clusterErr := self.doSyncOpToClusterAsync(true, coord, doLocalWrite, doLocalExit, doLocalCommit, doLocalRollback, | |||
There was a problem hiding this comment.
No async for cluster write, async is only for slave replication.
| @@ -116,8 +117,8 @@ func (self *NsqdCoordinator) PutMessageToCluster(topic *nsqd.Topic, | |||
| return false | |||
| } | |||
|
|
|||
| clusterErr := self.doSyncOpToCluster(true, coord, doLocalWrite, doLocalExit, doLocalCommit, doLocalRollback, | |||
| doRefresh, doSlaveSync, handleSyncResult) | |||
| clusterErr := self.doSyncOpToClusterAsync(true, coord, doLocalWrite, doLocalExit, doLocalCommit, doLocalRollback, | |||
There was a problem hiding this comment.
There is No async for cluster write, async is only for slave replication.
| if rpcErr == nil { | ||
| wg.Add(1) | ||
| id := nodeID | ||
| go func(){ |
There was a problem hiding this comment.
I think no goroutinues should be used here, just wait on the slave result channel is enough
| syncStart = time.Now() | ||
| } | ||
| select { | ||
| case <-time.After(c.c.RequestTimeout): |
There was a problem hiding this comment.
Can we use only one timeout for all slaves. Once timeout cancel all the slave replication and retry or just fail.
There was a problem hiding this comment.
Do we need onr timeout for all? My thought is that if all async rpc calls need wait, it only degrade to sync rpc call in time eclapse.
DoraALin
force-pushed
the
async-in-cluster-rpc-call
branch
from
2839dfc
to
b51d1f1
Compare
|
@absolute8511 for your review again. |
DoraALin
force-pushed
the
async-in-cluster-rpc-call
branch
from
b51d1f1
to
84a6757
Compare
|
@absolute8511 for your review |
Signed-off-by: 元守 <lulin@youzan.com> async in put messages to cluster Signed-off-by: 元守 <lulin@youzan.com> remove debug log Signed-off-by: 元守 <lulin@youzan.com> one doSyncOpToCluster Signed-off-by: 元守 <lulin@youzan.com>
DoraALin
force-pushed
the
async-in-cluster-rpc-call
branch
from
84a6757
to
32a31e0
Compare
|
@absolute8511 for your review. async rpc response wait incorporated in SlaveWriteResult.GetResult() |
| @@ -318,15 +320,12 @@ retrysync: | |||
| } | |||
| } | |||
| success = 0 | |||
| failedNodes = make(map[string]struct{}) | |||
There was a problem hiding this comment.
Why removing this? While we retry we need reset the failed info for all nodes
| var start time.Time | ||
| if checkCost { | ||
| start = time.Now() | ||
| } | ||
| // should retry if failed, and the slave should keep the last success write to avoid the duplicated | ||
| rpcErr = doSlaveSync(c, nodeID, tcData) | ||
| retSlave := doSlaveSync(c, nodeID, tcData) | ||
| rpcResps = append(rpcResps, RpcResp{nodeID, retSlave}) |
There was a problem hiding this comment.
rpcResps and totalTimeout should be reset while retry.
|
@DoraALin Could you please compare the detail benchmark before and after this changes. (both for benchmark test and the real server test.) |
|
@absolute8511 benchmark shows async rpc call does NOT fatser than replica synchronization. I will try with server test. Sync RPC call Async RPC call |
Signed-off-by: 元守 lulin@youzan.com
change rpc call in async way in PutMessageToCluster & PutMessagesToCluster
async in put messages to cluster