Cross validation with multiprocessing RuntimeError: dictionary changed size during iteration

I'm calling multiple runs with Optuna and automated script. No multiprocessing or threading used, only own Prophet call at cross-validation:
```python
df_cv = cross_validation(model, initial, period, horizon, parallel="processes")
```

Error happens very rarely, can't reproduce it.
Environment: Windows 10 (Docker at WSL2), Docker image `python:3.9.6-buster`, jupyter Lab server
Prophet version: v1.0

From logs we see that error happens at concatenating with pandas.concat:
```
return pd.concat(predicts, axis=0).reset_index(drop=True)
```

Exception log

```text
27.07 07:04:46: Checking study for item pasta and region 53
[I 2021-07-27 07:04:47,128] Using an existing study with name '53_pasta' instead of creating a new one.
27.07 07:04:47: Study already reached target with score 2.799. Tryin to improve...
27.07 07:04:47: Started optimization
[I 2021-07-27 07:04:55,733] Trial 8 finished with value: 12.502 and parameters: {'yeo_johnson_transform': True, 'growth': 'flat', 'changepoint_prior_scale': 0.04226910789745283, 'seasonality_mode': 'multiplicative', 'seasonality_prior_scale': 0.029436203771598793, 'holidays_prior_scale': 0.0782086726683088, 'weekly_seasonality': 6, 'yearly_seasonality': 194}. Best is trial 1 with value: 2.799.
[I 2021-07-27 07:05:27,546] Trial 10 finished with value: 11.055 and parameters: {'yeo_johnson_transform': False, 'growth': 'linear', 'changepoint_prior_scale': 0.35854747389934494, 'seasonality_mode': 'multiplicative', 'seasonality_prior_scale': 0.25591527344720655, 'holidays_prior_scale': 0.053195607802615744, 'weekly_seasonality': 29, 'yearly_seasonality': 177}. Best is trial 11 with value: 2.772.
[I 2021-07-27 07:10:14,416] Trial 13 finished with value: 2.809 and parameters: {'yeo_johnson_transform': True, 'growth': 'linear', 'changepoint_prior_scale': 0.0013082309862425263, 'seasonality_mode': 'multiplicative', 'seasonality_prior_scale': 0.010081785964888672, 'holidays_prior_scale': 1.4755212248457241, 'weekly_seasonality': 99, 'yearly_seasonality': 114}. Best is trial 11 with value: 2.772.
[I 2021-07-27 07:10:26,957] Trial 14 finished with value: 2.97 and parameters: {'yeo_johnson_transform': True, 'growth': 'linear', 'changepoint_prior_scale': 0.004983881761403154, 'seasonality_mode': 'multiplicative', 'seasonality_prior_scale': 0.02204438894059888, 'holidays_prior_scale': 0.8749377131606371, 'weekly_seasonality': 91, 'yearly_seasonality': 140}. Best is trial 12 with value: 2.77.
[I 2021-07-27 07:25:55,109] Trial 16 finished with value: 5.26 and parameters: {'yeo_johnson_transform': True, 'growth': 'linear', 'changepoint_prior_scale': 0.0011450757654519236, 'seasonality_mode': 'multiplicative', 'seasonality_prior_scale': 0.03867162200418062, 'holidays_prior_scale': 2.9152532236671456, 'weekly_seasonality': 100, 'yearly_seasonality': 78}. Best is trial 12 with value: 2.77.
[I 2021-07-27 07:26:04,749] Trial 19 finished with value: 2.787 and parameters: {'yeo_johnson_transform': True, 'growth': 'linear', 'changepoint_prior_scale': 0.009478868131939213, 'seasonality_mode': 'multiplicative', 'seasonality_prior_scale': 0.010265765014299938, 'holidays_prior_scale': 0.4408228489546783, 'weekly_seasonality': 77, 'yearly_seasonality': 142}. Best is trial 12 with value: 2.77.
[I 2021-07-27 07:26:13,496] Trial 20 finished with value: 3.006 and parameters: {'yeo_johnson_transform': True, 'growth': 'linear', 'changepoint_prior_scale': 0.0022465706864568586, 'seasonality_mode': 'multiplicative', 'seasonality_prior_scale': 0.5115529604430077, 'holidays_prior_scale': 0.9777520825028468, 'weekly_seasonality': 54, 'yearly_seasonality': 45}. Best is trial 12 with value: 2.77.
Exception in thread Thread-935:
Traceback (most recent call last):
File "/usr/local/lib/python3.9/threading.py", line 973, in _bootstrap_inner
self.run()
File "/usr/local/lib/python3.9/concurrent/futures/process.py", line 317, in run
result_item, is_broken, cause = self.wait_result_broken_or_wakeup()
File "/usr/local/lib/python3.9/concurrent/futures/process.py", line 376, in wait_result_broken_or_wakeup
worker_sentinels = [p.sentinel for p in self.processes.values()]
File "/usr/local/lib/python3.9/concurrent/futures/process.py", line 376, in
worker_sentinels = [p.sentinel for p in self.processes.values()]
RuntimeError: dictionary changed size during iteration

---- then I've stopped jupyterlab cell process ----

Process ForkProcess-1857:
Process ForkProcess-1858:
Traceback (most recent call last):
Traceback (most recent call last):
File "/usr/local/lib/python3.9/multiprocessing/process.py", line 315, in _bootstrap
self.run()
File "/usr/local/lib/python3.9/multiprocessing/process.py", line 315, in _bootstrap
self.run()
File "/usr/local/lib/python3.9/multiprocessing/process.py", line 108, in run
self._target(*self._args, **self._kwargs)
File "/usr/local/lib/python3.9/multiprocessing/process.py", line 108, in run
self._target(*self._args, **self._kwargs)
File "/usr/local/lib/python3.9/concurrent/futures/process.py", line 237, in _process_worker
call_item = call_queue.get(block=True)
File "/usr/local/lib/python3.9/concurrent/futures/process.py", line 237, in _process_worker
call_item = call_queue.get(block=True)
File "/usr/local/lib/python3.9/multiprocessing/queues.py", line 102, in get
with self._rlock:
File "/usr/local/lib/python3.9/multiprocessing/queues.py", line 103, in get
res = self._recv_bytes()
File "/usr/local/lib/python3.9/multiprocessing/synchronize.py", line 95, in __enter__
return self._semlock.__enter__()
File "/usr/local/lib/python3.9/multiprocessing/connection.py", line 221, in recv_bytes
buf = self._recv_bytes(maxlength)
KeyboardInterrupt
---------------------------------------------------------------------------
KeyboardInterrupt Traceback (most recent call last)
/tmp/ipykernel_8074/558810302.py in
----> 1 parallel_optimizations.run(
2 optuna_db_url = OPTUNA_DB_URL,
3 days_to_predict = 91,
4 sufficient_days = 30,
5 holidays = HOLIDAYS,

/tmp/ipykernel_8074/3921422868.py in run(self, optuna_db_url, days_to_predict, sufficient_days, holidays, all_combinations, cursor_min, cursor_max, stop_datetime, target_score, jobs_batch, batch_max_minutes, batch_improvement_expectation, study_max_minutes, verbose)
295 for item_name, region_num in self.all_combinations:
296 if not self.is_stop_datetime_exceeded(print_message=False):
--> 297 self.optimize_study(item_name, region_num, study_max_minutes, jobs_batch, batch_max_minutes, batch_improvement_expectation, try_improve=True)
298 else:
299 break

/tmp/ipykernel_8074/3921422868.py in optimize_study(self, item_name, region_num, study_max_minutes, jobs_batch, batch_max_minutes, batch_improvement_expectation, try_improve)
399 current_optimization_time = time.time()
400 try:
--> 401 study.optimize(target_func, n_trials=jobs_batch)
402 except (InsufficientData, ZeroData) as exc:
403 optimization_info = {

/usr/local/lib/python3.9/site-packages/optuna/study.py in optimize(self, func, n_trials, timeout, n_jobs, catch, callbacks, gc_after_trial, show_progress_bar)
399 )
400
--> 401 _optimize(
402 study=self,
403 func=func,

/usr/local/lib/python3.9/site-packages/optuna/_optimize.py in _optimize(study, func, n_trials, timeout, n_jobs, catch, callbacks, gc_after_trial, show_progress_bar)
63 try:
64 if n_jobs == 1:
---> 65 _optimize_sequential(
66 study,
67 func,

/usr/local/lib/python3.9/site-packages/optuna/_optimize.py in _optimize_sequential(study, func, n_trials, timeout, catch, callbacks, gc_after_trial, reseed_sampler_rng, time_start, progress_bar)
160
161 try:
--> 162 trial = _run_trial(study, func, catch)
163 except Exception:
164 raise

/usr/local/lib/python3.9/site-packages/optuna/_optimize.py in _run_trial(study, func, catch)
214
215 try:
--> 216 value_or_values = func(trial)
217 except exceptions.TrialPruned as e:
218 # TODO(mamu): Handle multi-objective cases.

/tmp/ipykernel_8074/3921422868.py in (trial)
382 print(f'{self.time_mark()}: Started optimization')
383
--> 384 target_func = lambda trial: IndividualOptimization.objective(
385 trial,
386 region_num=region_num,

/tmp/ipykernel_8074/3921422868.py in objective(trial, region_num, item_name, data_cache, days_to_predict, sufficient_days, holidays)
63 with suppress_stdout_stderr():
64 model.fit(train_df_transformed)
---> 65 df_cv = cross_validation(model, initial=initial, period=period, horizon=str(days_to_predict) + 'days', parallel="processes")
66
67 # Возвращаем результаты в масштаб исходных данных

/usr/local/lib/python3.9/site-packages/prophet/diagnostics.py in cross_validation(model, horizon, period, initial, parallel, cutoffs, disable_tqdm)
200
201 # Combine all predicted pd.DataFrame into one pd.DataFrame
--> 202 return pd.concat(predicts, axis=0).reset_index(drop=True)
203
204

/usr/local/lib/python3.9/site-packages/pandas/util/_decorators.py in wrapper(*args, **kwargs)
309 stacklevel=stacklevel,
310 )
--> 311 return func(*args, **kwargs)
312
313 return wrapper

/usr/local/lib/python3.9/site-packages/pandas/core/reshape/concat.py in concat(objs, axis, join, ignore_index, keys, levels, names, verify_integrity, sort, copy)
292 ValueError: Indexes have overlapping values: ['a']
293 """
--> 294 op = _Concatenator(
295 objs,
296 axis=axis,

/usr/local/lib/python3.9/site-packages/pandas/core/reshape/concat.py in __init__(self, objs, axis, join, keys, levels, names, ignore_index, verify_integrity, copy, sort)
346 objs = [objs[k] for k in keys]
347 else:
--> 348 objs = list(objs)
349
350 if len(objs) == 0:

/usr/local/lib/python3.9/concurrent/futures/process.py in _chain_from_iterable_of_lists(iterable)
557 careful not to keep references to yielded objects.
558 """
--> 559 for element in iterable:
560 element.reverse()
561 while element:

/usr/local/lib/python3.9/concurrent/futures/_base.py in result_iterator()
606 # Careful not to keep a reference to the popped future
607 if timeout is None:
--> 608 yield fs.pop().result()
609 else:
610 yield fs.pop().result(end_time - time.monotonic())

/usr/local/lib/python3.9/concurrent/futures/_base.py in result(self, timeout)
438 return self.__get_result()
439
--> 440 self._condition.wait(timeout)
441
442 if self._state in [CANCELLED, CANCELLED_AND_NOTIFIED]:

/usr/local/lib/python3.9/threading.py in wait(self, timeout)
310 try: # restore state no matter what (e.g., KeyboardInterrupt)
311 if timeout is None:
--> 312 waiter.acquire()
313 gotit = True
314 else:

KeyboardInterrupt:
File "/usr/local/lib/python3.9/multiprocessing/connection.py", line 419, in _recv_bytes
buf = self._recv(4)
File "/usr/local/lib/python3.9/multiprocessing/connection.py", line 384, in _recv
chunk = read(handle, remaining)
KeyboardInterrupt
```