Revert erasure coding attempt to fix bug

codex/codex.nim

@@ -11,7 +11,6 @@ import std/sequtils
 import std/strutils
 import std/os
 import std/tables
-import std/cpuinfo
 
 import pkg/chronos
 import pkg/presto
@@ -24,7 +23,6 @@ import pkg/stew/shims/net as stewnet
 import pkg/datastore
 import pkg/ethers except Rng
 import pkg/stew/io2
-import pkg/taskpools
 
 import ./node
 import ./conf
@@ -55,7 +53,6 @@ type
     codexNode: CodexNodeRef
     repoStore: RepoStore
     maintenance: BlockMaintainer
-    taskpool: Taskpool
 
   CodexPrivateKey* = libp2p.PrivateKey # alias
   EthWallet = ethers.Wallet
@@ -190,10 +187,6 @@ proc start*(s: CodexServer) {.async.} =
 proc stop*(s: CodexServer) {.async.} =
   notice "Stopping codex node"
 
-
-  s.taskpool.syncAll()
-  s.taskpool.shutdown()
-
   await allFuturesThrowing(
     s.restServer.stop(),
     s.codexNode.switch.stop(),
@@ -283,15 +276,12 @@ proc new*(
     else:
       none Prover
 
-    taskpool = Taskpool.new(num_threads = countProcessors())
-
     codexNode = CodexNodeRef.new(
       switch = switch,
       networkStore = store,
       engine = engine,
-      prover = prover,
       discovery = discovery,
-      taskpool = taskpool)
+      prover = prover)
 
     restServer = RestServerRef.new(
       codexNode.initRestApi(config, repoStore, config.apiCorsAllowedOrigin),
@@ -306,6 +296,4 @@ proc new*(
     config: config,
     codexNode: codexNode,
     restServer: restServer,
-    repoStore: repoStore,
+    repoStore: repoStore)
-    maintenance: maintenance,
-    taskpool: taskpool)

codex/erasure/erasure.nim

@@ -17,7 +17,6 @@ import std/sugar
 import pkg/chronos
 import pkg/libp2p/[multicodec, cid, multihash]
 import pkg/libp2p/protobuf/minprotobuf
-import pkg/taskpools
 
 import ../logutils
 import ../manifest
@@ -32,7 +31,6 @@ import ../errors
 import pkg/stew/byteutils
 
 import ./backend
-import ./asyncbackend
 
 export backend
 
@@ -73,7 +71,6 @@ type
     encoderProvider*: EncoderProvider
     decoderProvider*: DecoderProvider
     store*: BlockStore
-    taskpool: Taskpool
 
   EncodingParams = object
     ecK: Natural
@@ -295,23 +292,30 @@ proc encodeData(
       # TODO: Don't allocate a new seq every time, allocate once and zero out
       var
         data = seq[seq[byte]].new() # number of blocks to encode
+        parityData = newSeqWith[seq[byte]](params.ecM, newSeq[byte](manifest.blockSize.int))
 
       data[].setLen(params.ecK)
+      # TODO: this is a tight blocking loop so we sleep here to allow
+      # other events to be processed, this should be addressed
+      # by threading
+      await sleepAsync(10.millis)
 
       without resolved =?
         (await self.prepareEncodingData(manifest, params, step, data, cids, emptyBlock)), err:
           trace "Unable to prepare data", error = err.msg
           return failure(err)
 
-      trace "Erasure coding data", data = data[].len, parity = params.ecM
+      trace "Erasure coding data", data = data[].len, parity = parityData.len
 
-      without parity =? await asyncEncode(self.taskpool, encoder, data, manifest.blockSize.int, params.ecM), err:
+      if (
-        trace "Error encoding data", err = err.msg
+        let res = encoder.encode(data[], parityData);
-        return failure(err)
+        res.isErr):
+        trace "Unable to encode manifest!", error = $res.error
+        return failure($res.error)
 
       var idx = params.rounded + step
       for j in 0..<params.ecM:
-        without blk =? bt.Block.new(parity[j]), error:
+        without blk =? bt.Block.new(parityData[j]), error:
           trace "Unable to create parity block", err = error.msg
           return failure(error)
 
@@ -396,15 +400,21 @@ proc decode*(
   cids[].setLen(encoded.blocksCount)
   try:
     for step in 0..<encoded.steps:
+      # TODO: this is a tight blocking loop so we sleep here to allow
+      # other events to be processed, this should be addressed
+      # by threading
+      await sleepAsync(10.millis)
+      
       var
         data = seq[seq[byte]].new()
-        parity = seq[seq[byte]].new()
+        parityData = seq[seq[byte]].new()
+        recovered = newSeqWith[seq[byte]](encoded.ecK, newSeq[byte](encoded.blockSize.int))
 
       data[].setLen(encoded.ecK)    # set len to K
-      parity[].setLen(encoded.ecM)  # set len to M
+      parityData[].setLen(encoded.ecM)  # set len to M
 
       without (dataPieces, _) =?
-        (await self.prepareDecodingData(encoded, step, data, parity, cids, emptyBlock)), err:
+        (await self.prepareDecodingData(encoded, step, data, parityData, cids, emptyBlock)), err:
         trace "Unable to prepare data", error = err.msg
         return failure(err)
 
@@ -414,9 +424,11 @@ proc decode*(
 
       trace "Erasure decoding data"
 
-      without recovered =? await asyncDecode(self.taskpool, decoder, data, parity, encoded.blockSize.int), err:
+      if (
-        trace "Error decoding data", err = err.msg
+        let err = decoder.decode(data[], parityData[], recovered);
-        return failure(err)
+        err.isErr):
+        trace "Unable to decode data!", err = $err.error
+        return failure($err.error)
 
       for i in 0..<encoded.ecK:
         let idx = i * encoded.steps + step
@@ -470,13 +482,11 @@ proc new*(
   T: type Erasure,
   store: BlockStore,
   encoderProvider: EncoderProvider,
-  decoderProvider: DecoderProvider,
+  decoderProvider: DecoderProvider): Erasure =
-  taskpool: Taskpool): Erasure =
   ## Create a new Erasure instance for encoding and decoding manifests
   ##
 
   Erasure(
     store: store,
     encoderProvider: encoderProvider,
-    decoderProvider: decoderProvider,
+    decoderProvider: decoderProvider)
-    taskpool: taskpool)

codex/node.nim

@@ -13,7 +13,6 @@ import std/options
 import std/sequtils
 import std/strformat
 import std/sugar
-import std/cpuinfo
 import times
 
 import pkg/questionable
@@ -27,7 +26,6 @@ import pkg/libp2p/stream/bufferstream
 # TODO: remove once exported by libp2p
 import pkg/libp2p/routing_record
 import pkg/libp2p/signed_envelope
-import pkg/taskpools
 
 import ./chunker
 import ./slots
@@ -72,7 +70,6 @@ type
     contracts*: Contracts
     clock*: Clock
     storage*: Contracts
-    taskpool*: Taskpool
 
   CodexNodeRef* = ref CodexNode
 
@@ -214,7 +211,7 @@ proc fetchBatched*(
 proc streamSingleBlock(
   self: CodexNodeRef,
   cid: Cid
-): Future[?!LPstream] {.async.} =
+): Future[?!LPStream] {.async.} =
   ## Streams the contents of a single block.
   ##
   trace "Streaming single block", cid = cid
@@ -255,8 +252,7 @@ proc streamEntireDataset(
           erasure = Erasure.new(
             self.networkStore,
             leoEncoderProvider,
-            leoDecoderProvider,
+            leoDecoderProvider)
-            self.taskpool)
         without _ =? (await erasure.decode(manifest)), error:
           error "Unable to erasure decode manifest", manifestCid, exc = error.msg
           return failure(error)
@@ -442,8 +438,7 @@ proc setupRequest(
     erasure = Erasure.new(
       self.networkStore.localStore,
       leoEncoderProvider,
-      leoDecoderProvider,
+      leoDecoderProvider)
-      self.taskpool)
 
   without encoded =? (await erasure.encode(manifest, ecK, ecM)), error:
     trace "Unable to erasure code dataset"
@@ -779,8 +774,7 @@ proc new*(
   engine: BlockExcEngine,
   discovery: Discovery,
   prover = Prover.none,
-  contracts = Contracts.default,
+  contracts = Contracts.default): CodexNodeRef =
-  taskpool = Taskpool.new(num_threads = countProcessors())): CodexNodeRef =
   ## Create new instance of a Codex self, call `start` to run it
   ##
 
@@ -790,5 +784,4 @@ proc new*(
     engine: engine,
     prover: prover,
     discovery: discovery,
-    contracts: contracts,
+    contracts: contracts)
-    taskpool: taskpool)

tests/codex/node/helpers.nim

@@ -1,10 +1,8 @@
 import std/tables
 import std/times
-import std/cpuinfo
 
 import pkg/libp2p
 import pkg/chronos
-import pkg/taskpools
 import pkg/codex/codextypes
 import pkg/codex/chunker
 import pkg/codex/stores
@@ -83,7 +81,6 @@ template setupAndTearDown*() {.dirty.} =
     pendingBlocks: PendingBlocksManager
     discovery: DiscoveryEngine
     advertiser: Advertiser
-    taskpool: Taskpool
 
   let
     path = currentSourcePath().parentDir
@@ -113,14 +110,12 @@ template setupAndTearDown*() {.dirty.} =
     advertiser = Advertiser.new(localStore, blockDiscovery)
     engine = BlockExcEngine.new(localStore, wallet, network, discovery, advertiser, peerStore, pendingBlocks)
     store = NetworkStore.new(engine, localStore)
-    taskpool = Taskpool.new(num_threads = countProcessors())
     node = CodexNodeRef.new(
       switch = switch,
       networkStore = store,
       engine = engine,
       prover = Prover.none,
-      discovery = blockDiscovery,
+      discovery = blockDiscovery)
-      taskpool = taskpool)
 
   teardown:
     close(file)

tests/codex/node/testcontracts.nim

@@ -4,7 +4,6 @@ import std/math
 import std/times
 import std/sequtils
 import std/importutils
-import std/cpuinfo
 
 import pkg/chronos
 import pkg/stew/byteutils
@@ -15,7 +14,6 @@ import pkg/questionable/results
 import pkg/stint
 import pkg/poseidon2
 import pkg/poseidon2/io
-import pkg/taskpools
 
 import pkg/nitro
 import pkg/codexdht/discv5/protocol as discv5
@@ -82,7 +80,7 @@ asyncchecksuite "Test Node - Host contracts":
       manifestBlock = bt.Block.new(
         manifest.encode().tryGet(),
         codec = ManifestCodec).tryGet()
-      erasure = Erasure.new(store, leoEncoderProvider, leoDecoderProvider, taskpool)
+      erasure = Erasure.new(store, leoEncoderProvider, leoDecoderProvider)
 
     manifestCid = manifestBlock.cid
     manifestCidStr = $(manifestCid)

tests/codex/node/testnode.nim

@@ -4,7 +4,6 @@ import std/math
 import std/times
 import std/sequtils
 import std/importutils
-import std/cpuinfo
 
 import pkg/chronos
 import pkg/stew/byteutils
@@ -15,7 +14,6 @@ import pkg/questionable/results
 import pkg/stint
 import pkg/poseidon2
 import pkg/poseidon2/io
-import pkg/taskpools
 
 import pkg/nitro
 import pkg/codexdht/discv5/protocol as discv5
@@ -141,7 +139,7 @@ asyncchecksuite "Test Node - Basic":
 
   test "Setup purchase request":
     let
-      erasure = Erasure.new(store, leoEncoderProvider, leoDecoderProvider, taskpool)
+      erasure = Erasure.new(store, leoEncoderProvider, leoDecoderProvider)
       manifest = await storeDataGetManifest(localStore, chunker)
       manifestBlock = bt.Block.new(
         manifest.encode().tryGet(),

tests/codex/testerasure.nim

@@ -1,6 +1,5 @@
 import std/sequtils
 import std/sugar
-import std/cpuinfo
 
 import pkg/chronos
 import pkg/datastore
@@ -13,7 +12,6 @@ import pkg/codex/blocktype as bt
 import pkg/codex/rng
 import pkg/codex/utils
 import pkg/codex/indexingstrategy
-import pkg/taskpools
 
 import ../asynctest
 import ./helpers
@@ -28,7 +26,6 @@ suite "Erasure encode/decode":
   var manifest: Manifest
   var store: BlockStore
   var erasure: Erasure
-  var taskpool: Taskpool
   let repoTmp = TempLevelDb.new()
   let metaTmp = TempLevelDb.new()
 
@@ -39,8 +36,7 @@ suite "Erasure encode/decode":
     rng = Rng.instance()
     chunker = RandomChunker.new(rng, size = dataSetSize, chunkSize = BlockSize)
     store = RepoStore.new(repoDs, metaDs)
-    taskpool = Taskpool.new(num_threads = countProcessors())
+    erasure = Erasure.new(store, leoEncoderProvider, leoDecoderProvider)
-    erasure = Erasure.new(store, leoEncoderProvider, leoDecoderProvider, taskpool)
     manifest = await storeDataGetManifest(store, chunker)
 
   teardown: