TFF中clients_per_round超99时tff.templates.IterativeProcess.next报TypeError

问题描述 #

基于TensorFlow Federated（TFF）实现自定义联邦学习GAN训练循环，实现逻辑参考谷歌研究院开源的tff_gans.py代码。每轮训练的客户端数据通过如下代码获取：

相关代码 #

def client_dataset_fn():
    # Sample clients and data
    sampled_clients = np.random.choice(train_data.client_ids, size=cfg.clients_per_round, replace=False)
    datasets = [(next(client_gen_inputs_iterator),
                 train_data.create_tf_dataset_for_client(client_id).take(cfg.n_critic))
                for client_id in sampled_clients]
    return datasets

client_noise_inputs, client_real_data = zip(*client_dataset_fn())

错误现象 #

当cfg.clients_per_round参数设置为99及以下时，代码运行正常；当参数设置为100及更大值（总客户端数满足要求）时，抛出如下错误：

Traceback (most recent call last):
  File "main.py", line 109, in main
    metrics = run_single_trial(train_data, test_data, cfg)
  File "/mnt/workspace/tff/GAN/federated/fedgan_main.py", line 73, in run_single_trial
    metrics = train_loop(iterative_process, server_dataset_fn, client_dataset_fn, model, eval_hook_fn, cfg)
  File "/mnt/workspace/tff/GAN/federated/fedgan_main.py", line 124, in train_loop
    client_real_data)
  File "/usr/local/lib/python3.6/dist-packages/tensorflow_federated/python/core/impl/computation/function_utils.py", line 525, in __call__
    return context.invoke(self, arg)
  File "/usr/local/lib/python3.6/dist-packages/retrying.py", line 49, in wrapped_f
    return Retrying(*dargs, **dkw).call(f, *args, **kw)
  File "/usr/local/lib/python3.6/dist-packages/retrying.py", line 206, in call
    return attempt.get(self._wrap_exception)
  File "/usr/local/lib/python3.6/dist-packages/retrying.py", line 247, in get
    six.reraise(self.value[0], self.value[1], self.value[2])
  File "/usr/local/lib/python3.6/dist-packages/six.py", line 703, in reraise
    raise value
  File "/usr/local/lib/python3.6/dist-packages/retrying.py", line 200, in call
    attempt = Attempt(fn(*args, **kwargs), attempt_number, False)
  File "/usr/local/lib/python3.6/dist-packages/tensorflow_federated/python/core/impl/executors/execution_context.py", line 226, in invoke
    _ingest(executor, unwrapped_arg, arg.type_signature)))
  File "/usr/lib/python3.6/asyncio/base_events.py", line 484, in run_until_complete
    return future.result()
  File "/usr/local/lib/python3.6/dist-packages/tensorflow_federated/python/common_libs/tracing.py", line 396, in _wrapped
    return await coro
  File "/usr/local/lib/python3.6/dist-packages/tensorflow_federated/python/core/impl/executors/execution_context.py", line 111, in _ingest
    ingested = await asyncio.gather(*ingested)
  File "/usr/local/lib/python3.6/dist-packages/tensorflow_federated/python/core/impl/executors/execution_context.py", line 116, in _ingest
    return await executor.create_value(val, type_spec)
  File "/usr/local/lib/python3.6/dist-packages/tensorflow_federated/python/common_libs/tracing.py", line 201, in async_trace
    result = await fn(*fn_args, **fn_kwargs)
  File "/usr/local/lib/python3.6/dist-packages/tensorflow_federated/python/core/impl/executors/reference_resolving_executor.py", line 294, in create_value
    value, type_spec))
  File "/usr/local/lib/python3.6/dist-packages/tensorflow_federated/python/common_libs/tracing.py", line 201, in async_trace
    result = await fn(*fn_args, **fn_kwargs)
  File "/usr/local/lib/python3.6/dist-packages/tensorflow_federated/python/core/impl/executors/thread_delegating_executor.py", line 111, in create_value
    self._target_executor.create_value(value, type_spec))
  File "/usr/local/lib/python3.6/dist-packages/tensorflow_federated/python/core/impl/executors/thread_delegating_executor.py", line 105, in _delegate
    result_value = await _delegate_with_trace_ctx(coro, self._event_loop)
  File "/usr/local/lib/python3.6/dist-packages/tensorflow_federated/python/common_libs/tracing.py", line 396, in _wrapped
    return await coro
  File "/usr/local/lib/python3.6/dist-packages/tensorflow_federated/python/common_libs/tracing.py", line 201, in async_trace
    result = await fn(*fn_args, **fn_kwargs)
  File "/usr/local/lib/python3.6/dist-packages/tensorflow_federated/python/core/impl/executors/federating_executor.py", line 394, in create_value
    return await self._strategy.compute_federated_value(value, type_spec)
  File "/usr/local/lib/python3.6/dist-packages/tensorflow_federated/python/core/impl/executors/federated_composing_strategy.py", line 279, in compute_federated_value
    py_typecheck.check_type(value, list)
  File "/usr/local/lib/python3.6/dist-packages/tensorflow_federated/python/common_libs/py_typecheck.py", line 41, in check_type
    type_string(type_spec), type_string(type(target))))
TypeError: Expected list, found tuple.

调试发现报错指向的变量是上述代码生成的client_real_data和client_noise_inputs，二者是zip函数返回的tuple类型，该类型不会随每轮客户端数的设置发生变化，cfg.clients_per_round仅在随机采样客户端时被使用。目前无法解释该问题的触发原因。

依赖版本 #

• Python 3.6.9 或 3.8.10（两个版本均验证过）
• tensorflow 2.5.1
• tensorflow-federated 0.19.0
• retrying 1.3.3
• six 1.15.0

临时解决方案 #

通过list(tuple_var)手动将client_noise_inputs和client_real_data转换为列表类型可正常运行，但希望了解要求传入列表类型的根本原因。

内容的提问来源于stack exchange，提问作者Bjarne Pfitzner