New matching task flush buffer #2286
Conversation
* Add task flush buffer & UT * Add task queue manager
service/matching/db_task_manager.go
Outdated
| } | ||
| } | ||
|
|
||
| func (d *dbTaskManager) writeAppendTask( |
There was a problem hiding this comment.
why do you prefer appendTask over writeAppendTask?
There was a problem hiding this comment.
it is either write task or append task. writeAppend feels redundant.
service/matching/db_task_manager.go
Outdated
| return d.taskFlusher.appendTask(task) | ||
| } | ||
|
|
||
| func (d *dbTaskManager) readDispatchTask() { |
There was a problem hiding this comment.
this function actually read then dispatch a task, not simply doing in mem dispatch
| ) | ||
|
|
||
| type ( | ||
| dbTaskManager struct { |
There was a problem hiding this comment.
should be taskqueueManager or taskManager?
There was a problem hiding this comment.
no, this is managing the task which are going to be write to DB and read back
There was a problem hiding this comment.
But this is one instance per task queue. I'm trying to reduce new concept that people need to understand. Reuse the existing name would help folks who already understand current codebase.
There was a problem hiding this comment.
But this is one instance per task queue
this is per task queue, but managing write / read to DB.
I'm trying to reduce new concept that people need to understand.
this has nothing to do with implementation. implementation itself should be as clean as possible.
| return | ||
| } | ||
|
|
||
| err := d.dispatchTaskFn(newInternalTask( |
There was a problem hiding this comment.
if anything goes wrong, we will fall into a busy spin here?
There was a problem hiding this comment.
there is a for !d.isStopped() { at the beginning
d.finishTaskFn, will try to move this function from head of queue to end of queue (which i believe is wrong, but too many changes already)
plus, this (at least right now) only mimics the current behavior
There was a problem hiding this comment.
I think this needs to be addressed. We should avoid retry in busy loop. It can be a TODO with follow up PR.
There was a problem hiding this comment.
there will be following PRs
quite lots of places to optimize & simplify
but implementation (rewrite) should be done one step at a time
PRs already sent out / merged are focused on db task read / write
later there should be sync match related PRs & matching logic governing sync / async match policy
| return atomic.LoadInt32(&d.status) == common.DaemonStatusStopped | ||
| } | ||
|
|
||
| func (d *dbTaskManager) writerEventLoop() { |
There was a problem hiding this comment.
This writer loop logic belong to dbTaskWriter, same for reader loop belong to dbTaskReader. I think that would make code more organized.
There was a problem hiding this comment.
no, db reader / writer manages the internal state (acked task ID, next task ID for allocation) & actual read / write logic
the control logic exists here so signal read after task is persisted can be done easily
service/matching/db_task_manager.go
Outdated
| } | ||
| } | ||
|
|
||
| func (d *dbTaskManager) writeAppendTask( |
There was a problem hiding this comment.
it is either write task or append task. writeAppend feels redundant.
| return | ||
| } | ||
|
|
||
| err := d.dispatchTaskFn(newInternalTask( |
There was a problem hiding this comment.
I think this needs to be addressed. We should avoid retry in busy loop. It can be a TODO with follow up PR.
| ) | ||
|
|
||
| type ( | ||
| dbTaskManager struct { |
There was a problem hiding this comment.
But this is one instance per task queue. I'm trying to reduce new concept that people need to understand. Reuse the existing name would help folks who already understand current codebase.
service/matching/db_task_manager.go
Outdated
| return d.taskFlusher.appendTask(task) | ||
| } | ||
|
|
||
| func (d *dbTaskManager) readDispatchTask() { |
What changed?
Why?
Matching task queue logic cleanup
How did you test it?
New tests
Potential risks
N/A
Is hotfix candidate?
N/A