Integrate io threads

.gitmodules

@@ -193,3 +193,8 @@
 [submodule "vendor/npeg"]
 	path = vendor/npeg
 	url = https://github.com/zevv/npeg
+[submodule "vendor/threading"]
+	path = vendor/threading
+	url = https://github.com/nim-lang/threading.git
+	ignore = untracked
+	branch = master

codex/codex.nim

@@ -22,6 +22,7 @@ import pkg/nitro
 import pkg/stew/io2
 import pkg/stew/shims/net as stewnet
 import pkg/datastore
+import pkg/taskpools
 import pkg/ethers except Rng
 
 import ./node
@@ -216,9 +217,12 @@ proc new*(
       msg: "Unable to create discovery directory for block store: " & discoveryDir)
 
   let
-    discoveryStore = Datastore(
-      SQLiteDatastore.new(config.dataDir / CodexDhtProvidersNamespace)
-      .expect("Should create discovery datastore!"))
+    ioTp = Taskpool.new(200) # Some reasonable number of threads here
+    discoveryStore = ThreadDatastore
+      .new(ds = SQLiteDatastore.new(config.dataDir / CodexDhtProvidersNamespace)
+      .expect("Should create discovery datastore!"),
+      tp = ioTp)
+      .expect("Should create discovery datastore!")
 
     discovery = Discovery.new(
       switch.peerInfo.privateKey,
@@ -232,15 +236,24 @@ proc new*(
     network = BlockExcNetwork.new(switch)
 
     repoData = case config.repoKind
-                of repoFS: Datastore(FSDatastore.new($config.dataDir, depth = 5)
-                  .expect("Should create repo file data store!"))
-                of repoSQLite: Datastore(SQLiteDatastore.new($config.dataDir)
-                  .expect("Should create repo SQLite data store!"))
+                of repoFS: ThreadDatastore.new(
+                    ds = FSDatastore.new($config.dataDir, depth = 5)
+                    .expect("Should create repo file data store!"),
+                    tp = ioTp,
+                    withLocks = true)
+                    .expect("Should create threaded data store!")
+                of repoSQLite: ThreadDatastore.new(
+                    ds = SQLiteDatastore.new($config.dataDir / CodexRepoNamespace)
+                    .expect("Should create repo SQLite data store!"),
+                    tp = ioTp)
+                    .expect("Should create threaded data store!")
 
     repoStore = RepoStore.new(
       repoDs = repoData,
-      metaDs = SQLiteDatastore.new(config.dataDir / CodexMetaNamespace)
+      metaDs = ThreadDatastore.new(
+        ds = SQLiteDatastore.new(config.dataDir / CodexMetaNamespace)
         .expect("Should create meta data store!"),
+        tp = ioTp).expect("Should create threaded data store!"),
       quotaMaxBytes = config.storageQuota.uint,
       blockTtl = config.blockTtl)
 

tests/codex/stores/testrepostore.nim

@@ -16,6 +16,7 @@ import pkg/codex/chunker
 import pkg/codex/stores
 import pkg/codex/blocktype as bt
 import pkg/codex/clock
+import pkg/taskpools
 
 import ../helpers
 import ../helpers/mockclock
@@ -27,10 +28,12 @@ checksuite "Test RepoStore start/stop":
   var
     repoDs: Datastore
     metaDs: Datastore
+    tp: Taskpool
 
   setup:
-    repoDs = SQLiteDatastore.new(Memory).tryGet()
-    metaDs = SQLiteDatastore.new(Memory).tryGet()
+    tp = Taskpool.new(20)
+    repoDs = ThreadDatastore.new(ds = SQLiteDatastore.new(Memory).tryGet(), tp = tp).tryGet()
+    metaDs = ThreadDatastore.new(ds = SQLiteDatastore.new(Memory).tryGet(), tp = tp).tryGet()
 
   test "Should set started flag once started":
     let repo = RepoStore.new(repoDs, metaDs, quotaMaxBytes = 200)