Fix race condition of set_subtask_result

[chaokunyang]

What do these changes do?

• Fix set_subtask_result race condition since we release lock when _decref_input_subtasks.
• Fix duplicate subtasks submit.

Check code requirements

Related issue number

Closes #2814

• tests added / passed (if needed)
• Ensure all linting tests pass, see here for how to run them

[chaokunyang]

Seems there is a bug in worker_slot, see #2814

Traceback (most recent call last):
  File "mars/oscar/core.pyx", line 478, in mars.oscar.core._BaseActor.__on_receive__
    with debug_async_timeout('actor_lock_timeout',
  File "mars/oscar/core.pyx", line 481, in mars.oscar.core._BaseActor.__on_receive__
    async with self._lock:
  File "mars/oscar/core.pyx", line 482, in mars.oscar.core._BaseActor.__on_receive__
    result = func(*args, **kwargs)
  File "/Users/chaokunyang/ant/Development/DevProjects/python/mars/mars/services/scheduling/worker/workerslot.py", line 169, in release_free_slot
    assert acquired_slot_id == slot_id, f"acquired_slot_id {acquired_slot_id} != slot_id {slot_id}"
AssertionError: acquired_slot_id 1 != slot_id 0
2022-03-14 12:01:04,073 ERROR execution.py:120 -- Failed to run subtask PzeAnA7jYhspICgUe7eS81Vz on band numa-0
Traceback (most recent call last):
  File "/Users/chaokunyang/ant/Development/DevProjects/python/mars/mars/services/scheduling/worker/execution.py", line 331, in internal_run_subtask
    subtask_info.result = await self._retry_run_subtask(
  File "/Users/chaokunyang/ant/Development/DevProjects/python/mars/mars/services/scheduling/worker/execution.py", line 420, in _retry_run_subtask
    return await _retry_run(subtask, subtask_info, _run_subtask_once)
  File "/Users/chaokunyang/ant/Development/DevProjects/python/mars/mars/services/scheduling/worker/execution.py", line 107, in _retry_run
    raise ex
  File "/Users/chaokunyang/ant/Development/DevProjects/python/mars/mars/services/scheduling/worker/execution.py", line 67, in _retry_run
    return await target_async_func(*args)
  File "/Users/chaokunyang/ant/Development/DevProjects/python/mars/mars/services/scheduling/worker/execution.py", line 412, in _run_subtask_once
    await slot_manager_ref.release_free_slot(
  File "/Users/chaokunyang/ant/Development/DevProjects/python/mars/mars/oscar/backends/context.py", line 189, in send
    return self._process_result_message(result)
  File "/Users/chaokunyang/ant/Development/DevProjects/python/mars/mars/oscar/backends/context.py", line 70, in _process_result_message
    raise message.as_instanceof_cause()
  File "/Users/chaokunyang/ant/Development/DevProjects/python/mars/mars/oscar/backends/pool.py", line 542, in send
    result = await self._run_coro(message.message_id, coro)
  File "/Users/chaokunyang/ant/Development/DevProjects/python/mars/mars/oscar/backends/pool.py", line 333, in _run_coro
    return await coro
  File "/Users/chaokunyang/ant/Development/DevProjects/python/mars/mars/oscar/api.py", line 115, in __on_receive__
    return await super().__on_receive__(message)
  File "mars/oscar/core.pyx", line 506, in __on_receive__
    raise ex
  File "mars/oscar/core.pyx", line 478, in mars.oscar.core._BaseActor.__on_receive__
    with debug_async_timeout('actor_lock_timeout',
  File "mars/oscar/core.pyx", line 481, in mars.oscar.core._BaseActor.__on_receive__
    async with self._lock:
  File "mars/oscar/core.pyx", line 482, in mars.oscar.core._BaseActor.__on_receive__
    result = func(*args, **kwargs)
  File "/Users/chaokunyang/ant/Development/DevProjects/python/mars/mars/services/scheduling/worker/workerslot.py", line 169, in release_free_slot
    assert acquired_slot_id == slot_id, f"acquired_slot_id {acquired_slot_id} != slot_id {slot_id}"
AssertionError: [address=127.0.0.1:56418, pid=42775] acquired_slot_id 1 != slot_id 0

[wjsi]

LGTM

[hekaisheng]

LGTM