fix: separate ZstdDecompressor for each thread.

[jpivarski]

Does this fix the

>>> fut.compute(scheduler="synchronous")
<Array [[469, 459, 188, 17.8], ..., [414, ...]] type='203939 * var * float32'>
>>> fut.compute(scheduler="threads")
Traceback (most recent call last):
...
zstd.ZstdError: decompression error: Corrupted block detected

issue, @nsmith-?

[nsmith-]

I should have checked this sooner, but I get now

>>> events.MET.pt.compute(scheduler="synchronous")
<Array [78, 24.6, 5.05, 58.4, ..., 35.7, 29.9, 54] type='203939 * float32'>
>>> events.MET.pt.compute(scheduler="threads")
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/Users/ncsmith/src/coffea/.env/lib/python3.10/site-packages/dask/base.py", line 315, in compute
    (result,) = compute(self, traverse=False, **kwargs)
  File "/Users/ncsmith/src/coffea/.env/lib/python3.10/site-packages/dask/base.py", line 600, in compute
    results = schedule(dsk, keys, **kwargs)
  File "/Users/ncsmith/src/coffea/.env/lib/python3.10/site-packages/dask/threaded.py", line 89, in get
    results = get_async(
  File "/Users/ncsmith/src/coffea/.env/lib/python3.10/site-packages/dask/local.py", line 511, in get_async
    raise_exception(exc, tb)
  File "/Users/ncsmith/src/coffea/.env/lib/python3.10/site-packages/dask/local.py", line 319, in reraise
    raise exc
  File "/Users/ncsmith/src/coffea/.env/lib/python3.10/site-packages/dask/local.py", line 224, in execute_task
    result = _execute_task(task, data)
  File "/Users/ncsmith/src/coffea/.env/lib/python3.10/site-packages/dask/core.py", line 119, in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
  File "/Users/ncsmith/src/coffea/.env/lib/python3.10/site-packages/dask/optimization.py", line 990, in __call__
    return core.get(self.dsk, self.outkey, dict(zip(self.inkeys, args)))
  File "/Users/ncsmith/src/coffea/.env/lib/python3.10/site-packages/dask/core.py", line 149, in get
    result = _execute_task(task, cache)
  File "/Users/ncsmith/src/coffea/.env/lib/python3.10/site-packages/dask/core.py", line 119, in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
  File "/Users/ncsmith/src/coffea/.env/lib/python3.10/site-packages/uproot/_dask.py", line 592, in __call__
    return self.ttrees[i].arrays(
  File "/Users/ncsmith/src/coffea/.env/lib/python3.10/site-packages/uproot/behaviors/TBranch.py", line 834, in arrays
    _ranges_or_baskets_to_arrays(
  File "/Users/ncsmith/src/coffea/.env/lib/python3.10/site-packages/uproot/behaviors/TBranch.py", line 3124, in _ranges_or_baskets_to_arrays
    uproot.source.futures.delayed_raise(*obj)
  File "/Users/ncsmith/src/coffea/.env/lib/python3.10/site-packages/uproot/source/futures.py", line 36, in delayed_raise
    raise exception_value.with_traceback(traceback)
  File "/Users/ncsmith/src/coffea/.env/lib/python3.10/site-packages/uproot/behaviors/TBranch.py", line 3037, in chunk_to_basket
    basket = uproot.models.TBasket.Model_TBasket.read(
  File "/Users/ncsmith/src/coffea/.env/lib/python3.10/site-packages/uproot/model.py", line 867, in read
    self.read_members(chunk, cursor, context, file)
  File "/Users/ncsmith/src/coffea/.env/lib/python3.10/site-packages/uproot/models/TBasket.py", line 291, in read_members
    uncompressed = uproot.compression.decompress(
  File "/Users/ncsmith/src/coffea/.env/lib/python3.10/site-packages/uproot/compression.py", line 415, in decompress
    uncompressed_bytestring = decompressor.decompress(
  File "/Users/ncsmith/src/coffea/.env/lib/python3.10/site-packages/uproot/compression.py", line 249, in decompress
    return self.decompressor.decompress(data)
  File "/Users/ncsmith/src/coffea/.env/lib/python3.10/site-packages/uproot/compression.py", line 243, in decompressor
    if self._decompressor.obj is None:
AttributeError: '_thread._local' object has no attribute 'obj'

The issue is that, apparently, each thread has an independent attribute dictionary:

>>> import threading
>>> x = threading.local()
>>> x.obj = None
>>> print(hasattr(x, "obj"))
True
>>> t = threading.Thread(target=lambda x: print(hasattr(x, "obj")), args=(x,))
>>> t.start()
False

[nsmith-]

I'll prepare a new PR