Errors in multithreading

[hwei-hw]

Thanks for sharing this great work!

When I try to deploy it on my machine, I encounter one error about the multithreading in line 274 to line 279 at cllm/services/tog/tool.py:

results = batch_exec(
send_message,
arguments_list,
min(15, len(arguments_list)),
await_result=True,
)

When I submit who are you? in the website, the parameter min(15, len(arguments_list)) is 1. Then, I encounter this the following error:

Traceback (most recent call last):
File "/home/haowei/anaconda3/envs/cllm/lib/python3.10/multiprocessing/queues.py", line 244, in _feed
obj = _ForkingPickler.dumps(obj)
File "/home/haowei/anaconda3/envs/cllm/lib/python3.10/multiprocessing/reduction.py", line 51, in dumps
cls(buf, protocol).dump(obj)
TypeError: cannot pickle '_thread.RLock' object
The above exception was the direct cause of the following exception:
Traceback (most recent call last):
File "/home/haowei/anaconda3/envs/cllm/lib/python3.10/site-packages/uvicorn/protocols/http/httptools_impl.py", line 426, in run_asgi
result = await app( # type: ignore[func-returns-value]
File "/home/haowei/anaconda3/envs/cllm/lib/python3.10/site-packages/uvicorn/middleware/proxy_headers.py", line 84, in call
return await self.app(scope, receive, send)
File "/home/haowei/anaconda3/envs/cllm/lib/python3.10/site-packages/fastapi/applications.py", line 1106, in call
await super().call(scope, receive, send)
File "/home/haowei/anaconda3/envs/cllm/lib/python3.10/site-packages/starlette/applications.py", line 122, in call
await self.middleware_stack(scope, receive, send)
File "/home/haowei/anaconda3/envs/cllm/lib/python3.10/site-packages/starlette/middleware/errors.py", line 184, in call
raise exc
File "/home/haowei/anaconda3/envs/cllm/lib/python3.10/site-packages/starlette/middleware/errors.py", line 162, in call
await self.app(scope, receive, _send)
File "/home/haowei/anaconda3/envs/cllm/lib/python3.10/site-packages/starlette/middleware/exceptions.py", line 79, in call
raise exc
File "/home/haowei/anaconda3/envs/cllm/lib/python3.10/site-packages/starlette/middleware/exceptions.py", line 68, in call
await self.app(scope, receive, sender)
File "/home/haowei/anaconda3/envs/cllm/lib/python3.10/site-packages/fastapi/middleware/asyncexitstack.py", line 20, in call
raise e
File "/home/haowei/anaconda3/envs/cllm/lib/python3.10/site-packages/fastapi/middleware/asyncexitstack.py", line 17, in call
await self.app(scope, receive, send)
File "/home/haowei/anaconda3/envs/cllm/lib/python3.10/site-packages/starlette/routing.py", line 718, in call
await route.handle(scope, receive, send)
File "/home/haowei/anaconda3/envs/cllm/lib/python3.10/site-packages/starlette/routing.py", line 276, in handle
await self.app(scope, receive, send)
File "/home/haowei/anaconda3/envs/cllm/lib/python3.10/site-packages/starlette/routing.py", line 66, in app
response = await func(request)
File "/home/haowei/anaconda3/envs/cllm/lib/python3.10/site-packages/fastapi/routing.py", line 274, in app
raw_response = await run_endpoint_function(
File "/home/haowei/anaconda3/envs/cllm/lib/python3.10/site-packages/fastapi/routing.py", line 191, in run_endpoint_function
return await dependant.call(**values)
File "/home/haowei/project/ControlLLM/cllm/services/pool.py", line 19, in innner_wrapper
return await func(*args, **kwargs)
File "/home/haowei/project/ControlLLM/cllm/services/tog/launch.py", line 57, in tog
output = model(request, subtasks)
File "/home/haowei/project/ControlLLM/cllm/services/tog/launch.py", line 46, in call
return self.got.solve(request, subtasks, multi_processing)
File "/home/haowei/project/ControlLLM/cllm/services/tog/tool.py", line 1105, in solve
optimal_solutions_list = batch_exec(
File "/home/haowei/project/ControlLLM/cllm/services/tog/tool.py", line 119, in batch_exec
return _exec(func, arguments_list)
File "/home/haowei/project/ControlLLM/cllm/services/tog/tool.py", line 97, in _exec
results.append(func(*arguments))
File "/home/haowei/project/ControlLLM/cllm/services/tog/tool.py", line 274, in solve
results = batch_exec(
File "/home/haowei/project/ControlLLM/cllm/services/tog/tool.py", line 121, in batch_exec
return _async_exec(func, arguments_list, max_workers, await_result)
File "/home/haowei/project/ControlLLM/cllm/services/tog/tool.py", line 112, in _async_exec
results[i] = r.result()
File "/home/haowei/anaconda3/envs/cllm/lib/python3.10/concurrent/futures/_base.py", line 451, in result
return self.__get_result()
File "/home/haowei/anaconda3/envs/cllm/lib/python3.10/concurrent/futures/_base.py", line 403, in __get_result
raise self._exception
File "/home/haowei/anaconda3/envs/cllm/lib/python3.10/multiprocessing/queues.py", line 244, in _feed
obj = _ForkingPickler.dumps(obj)
File "/home/haowei/anaconda3/envs/cllm/lib/python3.10/multiprocessing/reduction.py", line 51, in dumps
cls(buf, protocol).dump(obj)
TypeError: cannot pickle '_thread.RLock' object
^CINFO: Shutting down
INFO: Waiting for application shutdown.
INFO: Application shutdown complete.
INFO: Finished server process [860450]
Process finished with exit code 0

But when I set this parameter as 0, which means:

results = batch_exec(
send_message,
arguments_list,
0,
await_result=True,
)

The whole code work! So, maybe there are some bugs in this multithreading code?

Looking forward your reply.

[liu-zhy]

Thanks for your feedback! I will check and fix it ASAP.