feat: thread pool #40
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change | ||||||
|---|---|---|---|---|---|---|---|---|
|
|
@@ -5,7 +5,7 @@ | |||||||
| when defined(linux): | ||||||||
| {.passl: "-Wl,-soname,libsds.so".} | ||||||||
|
|
||||||||
| import std/[typetraits, tables, atomics, locks], chronos, chronicles | ||||||||
| import | ||||||||
| ./sds_thread/sds_thread, | ||||||||
| ./alloc, | ||||||||
|
|
@@ -57,6 +57,29 @@ template callEventCallback(ctx: ptr SdsContext, eventName: string, body: untyped | |||||||
| RET_ERR, unsafeAddr msg[0], cast[csize_t](len(msg)), ctx[].eventUserData | ||||||||
| ) | ||||||||
|
|
||||||||
| var | ||||||||
| ctxPool: seq[ptr SdsContext] | ||||||||
|
There was a problem hiding this comment.
Suggested change
something like this - assuming you don't want more than 32 concurrent contexts
Collaborator
Author
There was a problem hiding this comment. I think it's okay for now to allow as much as needed.
Collaborator
Author
There was a problem hiding this comment. I don't know how many is needed. And eventually we'll need a proper fix anyway There was a problem hiding this comment. no, I mean using a seq here is undefined behavior - if it grows more than once, it'll crash - seqs are tied to a particular thread, arrays are not - for a cross-thread seq, one needs to use
Collaborator
Author
There was a problem hiding this comment. We tested this to work for now in Go tests and Status App. |
||||||||
| ctxPoolLock: Lock | ||||||||
|
|
||||||||
| proc acquireCtx(callback: SdsCallBack, userData: pointer): ptr SdsContext = | ||||||||
| ctxPoolLock.acquire() | ||||||||
| defer: ctxPoolLock.release() | ||||||||
| if ctxPool.len > 0: | ||||||||
| result = ctxPool.pop() | ||||||||
| else: | ||||||||
| result = sds_thread.createSdsThread().valueOr: | ||||||||
| let msg = "Error in createSdsThread: " & $error | ||||||||
| callback(RET_ERR, unsafeAddr msg[0], cast[csize_t](len(msg)), userData) | ||||||||
| return nil | ||||||||
|
|
||||||||
| proc releaseCtx(ctx: ptr SdsContext) = | ||||||||
| ctxPoolLock.acquire() | ||||||||
| defer: ctxPoolLock.release() | ||||||||
| ctx.userData = nil | ||||||||
| ctx.eventCallback = nil | ||||||||
| ctx.eventUserData = nil | ||||||||
| ctxPool.add(ctx) | ||||||||
|
|
||||||||
| proc handleRequest( | ||||||||
| ctx: ptr SdsContext, | ||||||||
| requestType: RequestType, | ||||||||
|
|
@@ -140,10 +163,9 @@ proc SdsNewReliabilityManager( | |||||||
| echo "error: missing callback in NewReliabilityManager" | ||||||||
| return nil | ||||||||
|
|
||||||||
| ## Create or reuse the SDS thread that will keep waiting for req from the main thread. | ||||||||
| var ctx = acquireCtx(callback, userData) | ||||||||
| if ctx.isNil(): | ||||||||
| return nil | ||||||||
|
|
||||||||
| ctx.userData = userData | ||||||||
|
|
@@ -183,14 +205,20 @@ proc SdsCleanupReliabilityManager( | |||||||
| initializeLibrary() | ||||||||
| checkLibsdsParams(ctx, callback, userData) | ||||||||
|
|
||||||||
| let resetRes = handleRequest( | ||||||||
| ctx, | ||||||||
| RequestType.LIFECYCLE, | ||||||||
| SdsLifecycleRequest.createShared(SdsLifecycleMsgType.RESET_RELIABILITY_MANAGER), | ||||||||
| callback, | ||||||||
| userData, | ||||||||
| ) | ||||||||
|
|
||||||||
| if resetRes == RET_ERR: | ||||||||
| return RET_ERR | ||||||||
|
|
||||||||
| releaseCtx(ctx) | ||||||||
|
|
||||||||
| # handleRequest already invoked the callback; nothing else to signal here. | ||||||||
| return RET_OK | ||||||||
|
|
||||||||
| proc SdsResetReliabilityManager( | ||||||||
|
|
||||||||
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
seqcannot be used cross-thread until we migrate to the orc memory manager meaning that this list needs to be managed manually - the absolutely easiest way to do that is to use anarray+ counter and limit the number of sds contexts that can be created.