class_docstr: Reduce
class_docstr_tok: ["Reduce"]
code: def test_multithread_get(self):
      def reduce_op(*seqs):
          def func(seq):
              ret = 0
              pre_i = -1
              for i in seq:
                  assert pre_i < i
                  pre_i = i
                  ret += i
              return ret
  
          pool = ThreadPoolExecutor()
          fs = []
          ret = []
          for seq in seqs:
              f = pool.submit(func, seq)
              fs.append(f)
          for f in fs:
              ret.append(f.result())
          return ret[:2]
  
      in_que = DataQueue(
          [
              ("num1", ColumnType.QUEUE),
              ("num2", ColumnType.QUEUE),
              ("num3", ColumnType.QUEUE),
              ("num4", ColumnType.QUEUE),
              ("side_by", ColumnType.QUEUE),
          ],
          max_size=-1,
      )
  
      out_que = DataQueue(
          [
              ("sum1", ColumnType.SCALAR),
              ("sum2", ColumnType.SCALAR),
              ("side_by", ColumnType.QUEUE),
          ]
      )
  
      node_info = {
          "inputs": ("num1", "num2"),
          "outputs": ("sum1", "sum2"),
          "op_info": {
              "type": "lambda",
              "operator": reduce_op,
              "tag": "main",
              "init_args": None,
              "init_kws": {},
          },
          "iter_info": {"type": "reduce", "param": {}},
          "config": {"name": "test"},
          "next_nodes": ["_output"],
      }
  
      num_size = 10000
      node = create_node(
          NodeRepr(uid="test_node", **node_info),
          self.op_pool,
          [in_que],
          [out_que],
      )
      self.assertTrue(node.initialize())
      f = self.thread_pool.submit(node.process)
      for i in range(num_size):
          in_que.put([i] * 5)
      in_que.seal()
      f.result()
      self.assertTrue(node.status == NodeStatus.FINISHED)
      self.assertTrue(out_que.sealed)
      ret = out_que.to_list()
      self.assertEqual(len(ret), num_size)
      side_by = [item[2] for item in ret]
  
      def check(x, y):
          assert x < y
          return 0
  
      reduce(check, side_by)
      self.assertEqual(ret[0][0], 49995000)
      self.assertEqual(ret[0][1], 49995000)
  
imports: import unittest
  import time
  import copy
  import uuid
  from functools import reduce
  from concurrent.futures import ThreadPoolExecutor
  from towhee.runtime.node_repr import NodeRepr
  from towhee.runtime.nodes import create_node, NodeStatus
  from towhee.runtime.data_queue import DataQueue, ColumnType, Empty
  from towhee.runtime.operator_manager import OperatorPool
label_desc: Use image node_id 9 for calling the TestReduce obj's underlying member method code with example usage: obj.test_multithread_get() and returns: ret, int, ret