Unknown type torch.Tensor, caused by mismatching decomposers

[falibabaei]

Hi,

I have changed my code using the following example:
https://github.com/NVIDIA/NVFlare/blob/main/examples/hello-world/ml-to-fl/pt/code/cifar10_structured_fl.py
I have installed the latest version (=2.4.0rc5). Now I get the error message

• FederatedClient - INFO - Starting to push execute result.
  2023-11-23 15:55:29,897 - nvflare.fuel.f3.streaming.byte_receiver - ERROR - Rx[SID:1700775166009518 from site-1.simulate_job for server_command/submit_update] callback <bound method BlobHandler.handle_blob_cb of <nvflare.fuel.f3.streaming.blob_streamer.BlobHandler object at 0x7f8e043e4400>> throws exception: Unknown type torch.Tensor, caused by mismatching decomposers
  The code for training is :

  def train_model( input_model=None, epochs=2 , lr= 0.001 ):
  """Return the trained model and train/test accuracy/loss"""
  #if not do_training:
  # return None, None
  model.load_state_dict(input_model.params)
  optimizer=optim.Adam(model.parameters(), lr=lr)

    history = {'train_loss': [], 'train_acc': [], 'test_loss': [], 'test_acc': []}
    for e in range(1, epochs + 1):
        print('Epoch {}/{}'.format(e, epochs))
        for phase in ['train', 'test']:
            if phase == 'train':  """Return the trained model and train/test accuracy/loss"""
    #if not do_training:
    #    return None, None
    model.load_state_dict(input_model.params)
    optimizer=optim.Adam(model.parameters(), lr=lr)
  
    history = {'train_loss': [], 'train_acc': [], 'test_loss': [], 'test_acc': []}
    for e in range(1, epochs + 1):
        print('Epoch {}/{}'.format(e, epochs))
        for phase in ['train', 'test']:
            if phase == 'train':
                model.train()  # set model to training mode for training phase
            else:
                model.eval()  # set model to evaluation mode for test phase
  
            running_loss = 0.0  # record the training/test loss for each epoch
            running_corrects = 0  # record the number of correct predicts by the model for each epoch
  
            for features, labels in image_dataloaders[phase]:
                # send data to gpu if possible
                features = features.to(device)
                labels = labels.to(device)
  
                # reset the parameter gradients after each batch to avoid double-counting
                optimizer.zero_grad()
  
                # forward pass
                # set parameters to be trainable only at training phase
                with torch.set_grad_enabled(phase == 'train'):
                    outcomes = model(features)
                    pred_labels = outcomes.round()  # round up forward outcomes to get predicted labels
                    labels = labels.unsqueeze(1).type(torch.float)
                    loss = loss_func(outcomes, labels)  # calculate loss
  
                    # backpropagation only for training phase
                    if phase == 'train':
                        loss.backward()
                        optimizer.step()
  
                # record loss and correct predicts of each bach
                running_loss += loss.item() * features.size(0)
                running_corrects += torch.sum(pred_labels == labels.data)
  
            # record loss and correct predicts of each epoch and stored in history
            epoch_loss = running_loss / len(image_datasets[phase])
            epoch_acc = running_corrects.double() / len(image_datasets[phase])
  
    torch.save(model.state_dict(), PATH)
    # (4) construct trained FL model
    output_model = flare.FLModel(params=model.cpu().state_dict())
    return output_model
  

'''
I appreciate your help

[YuanTingHsieh]

@falibabaei can you paste your config_fed_server and config_fed_client for this job ?

[falibabaei]

I used the job template sag and only changed the name of the network.
'''
nvflare job create -force -j ./jobs/decorator -w sag_pt -sd

./jobs/client_api/app/custom -f config_fed_client.conf app_script=brain_tumor_client.py -f config_fed_server.conf model_class_path=net.TumorNet
'''
I had to install NVFlare version is 2.4.0rc5+38.g79de070c.dirty, because with 2.4.0rc5 I can not even execute CIFAR10 example.
One problem I thought of was the validation of the test set after each epoch. So I removed the validation but I still encourage the same error. Here is my log
log.txt

@flare.train
def train_model( input_model=None,  epochs=2 , lr= 0.001 ):
    """Return the trained model and train/test accuracy/loss"""
    #if not do_training:
    #    return None, None
    model.load_state_dict(input_model.params)
    optimizer=optim.Adam(model.parameters(), lr=lr)

   
    for e in range(1, epochs + 1):
        print('Epoch {}/{}'.format(e, epochs))
        

        running_loss = 0.0  # record the training/test loss for each epoch
        running_corrects = 0  # record the number of correct predicts by the model for each epoch

        for features, labels in train_dataloader:
                # send data to gpu if possible
                features = features.to(device)
                labels = labels.to(device)

                # reset the parameter gradients after each batch to avoid double-counting
                optimizer.zero_grad()

                # forward pass
                # set parameters to be trainable only at training phase
               
                outcomes = model(features)
                pred_labels = outcomes.round()  # round up forward outcomes to get predicted labels
                labels = labels.unsqueeze(1).type(torch.float)
                loss = loss_func(outcomes, labels)  # calculate loss

                    # backpropagation only for training phase
                
                loss.backward()
                optimizer.step()

                # record loss and correct predicts of each bach
                running_loss += loss.item() * features.size(0)
                running_corrects += torch.sum(pred_labels == labels.data)

            # record loss and correct predicts of each epoch and stored in history
        epoch_loss = running_loss / len(train_dataloader)
        epoch_acc = running_corrects.double() / len(train_dataloader)

        print('Loss: {:.4f} Acc: {:.4f}'.format( epoch_loss, epoch_acc))
            
    torch.save(model.state_dict(), PATH)
    # (4) construct trained FL model
    output_model = flare.FLModel(params=model.cpu().state_dict())
    # (8) send model back to NVFlare
    #flare.send(output_model)
    return output_model

def  evaluate(input_weights=None):
    """Return the trained model and train/test accuracy/loss"""
    #if not do_training:
    #    return None, None
    model= TumorNet()
    model.load_state_dict(input_weights)
    model.eval()  # set model to evaluation mode for test phase
    running_corrects = 0 
    for features, labels in valid_dataloader:
                # send data to gpu if possible
            features = features.to(device)
            labels = labels.to(device)

            with torch.no_grad():
                outcomes = model(features)
                pred_labels = outcomes.round()  # round up forward outcomes to get predicted labels
                labels = labels.unsqueeze(1).type(torch.float)
        
                running_corrects += torch.sum(pred_labels == labels.data)

            # record loss and correct predicts of each epoch and stored in history
            acc = running_corrects.double() / len(valid_dataloader)
            

    return 100 * acc

# (5) decorates with flare.evaluate and load model from the first argument
@flare.evaluate
def fl_evaluate(input_model=None):
    return evaluate(input_weights=input_model.params)


    # (6) receives FLModel from NVFlare
input_model = flare.receive()
print(f"current_round={input_model.current_round}")
                # (7) call fl_evaluate method before training
                #       to evaluate on the received/aggregated model
global_metric = fl_evaluate(input_model)
print(f"Accuracy of the global model on the test images: {global_metric} %")
                # call train method
            
output_model=train_model(input_model, epochs=3, lr=0.001) 
#flare.send(output_model)
metric = evaluate(input_weights=torch.load(PATH))
print(f"Accuracy of the trained model on the test images: {metric} %")

MY CLIENT CONFIG:

format_version = 2
app_script = "brain_tumor_client.py"
app_config = ""
executors = [
{
tasks = [
"train"
]
executor {
path = "nvflare.app_opt.pt.client_api_launcher_executor.PTClientAPILauncherExecutor"
args {
launcher_id = "launcher"
pipe_id = "pipe"
heartbeat_timeout = 60
params_exchange_format = "pytorch"
params_transfer_type = "DIFF"
train_with_evaluation = true
launch_once = true
}
}
}
]
task_data_filters = []
task_result_filters = []
components = [
{
id = "launcher"
path = "nvflare.app_common.launchers.subprocess_launcher.SubprocessLauncher"
args {
script = "python3 custom/{app_script} {app_config} "
}
}
{
id = "pipe"
path = "nvflare.fuel.utils.pipe.file_pipe.FilePipe"
args {
mode = "PASSIVE"
root_path = ""
}
}
]

MYSERVER CONFIG:

format_version = 2
task_data_filters = []
task_result_filters = []
model_class_path = "net.TumorNet"
workflows = [
{
id = "scatter_and_gather"
path = "nvflare.app_common.workflows.scatter_and_gather.ScatterAndGather"
args {
min_clients = 2
num_rounds = 4
start_round = 0
wait_time_after_min_received = 0
aggregator_id = "aggregator"
persistor_id = "persistor"
shareable_generator_id = "shareable_generator"
train_task_name = "train"
train_timeout = 0
}
}
]
components = [
{
id = "persistor"
path = "nvflare.app_opt.pt.file_model_persistor.PTFileModelPersistor"
args {
model {
path = "{model_class_path}"
}
}
}
{
id = "shareable_generator"
path = "nvflare.app_common.shareablegenerators.full_model_shareable_generator.FullModelShareableGenerator"
args {}
}
{
id = "aggregator"
path = "nvflare.app_common.aggregators.intime_accumulate_model_aggregator.InTimeAccumulateWeightedAggregator"
args {
expected_data_kind = "WEIGHT_DIFF"
}
}
{
id = "model_selector"
path = "nvflare.app_common.widgets.intime_model_selector.IntimeModelSelector"
args {
key_metric = "accuracy"
}
}
]

[chesterxgchen]

@YuanTingHsieh can you look into this ?

[chesterxgchen]

Looking at the log, looks like error is due to mismatach decomposers. is it possible due to you have two different version of NVFLARE where the decomposer has changed from one version to another.

Let's try this:

• uninstall all nvflare versions
• pick the version you like to work with ( latest from main branch or RC release version), install that version, and see if you still experience this problem

[falibabaei]

Hi,
With the same env, I can run the cifar_10 example. I just can not see any difference between my code and the cifar_10 example.

[chesterxgchen]

ok, so the cifar_10 example works, with your code it doesn't work. What's the error ? is still the decomposition error you reported before ?

[falibabaei]

yes, it is still the same error

[falibabaei]

Hi,
I found out that the error cames from the evaluation function!! I used torch.sum to calculate the accuracy and it seems that this should not be the case
Thanks for your help :)
Still, there is a small problem. I can not see my training logs and prints in the log.txt. The model is trained and saved, but the log does not show any of my prints during training
log.txt

[nvidianz]

@falibabaei I am puzzled by the decomposer error. We registered the decomposer for the PyTorch tensor in simulator so that error shouldn't occur in the same process. Do you spawn another process or alter PYTHONPATH? It would be great if you can share your code. Thanks!

[falibabaei]

Hi,
No, I did not change the Python path.
Here is my code. https://github.com/falibabaei/brain_tumor_detection_nvflare/tree/main
I am using the code in code/tumor_detection_decorator.py

[YuanTingHsieh]

@falibabaei the logs of external process (launched by SubprocessLauncher) is not captured by NVFlare now, we are working to enhance that.

To your finding, we will also try on our side to return a torch Tensor for evaluate method and see if we encounter the same issue.

[falibabaei]

For the cifar_10 example, I can see the log of the training in the terminal and in the train_logfile for each site
train_logfile.txt
but for my example this is not the case. I can not see my training log in the terminal or in the train_logfile for each site.
train_logfile.txt

[YuanTingHsieh]

@falibabaei thanks for the logs, how do you run your system?
Are you using real-world deployment or simulator?

Have you tried running the example following https://github.com/NVIDIA/NVFlare/tree/main/examples/hello-world/ml-to-fl/pt#the-decorator-use-case ?
Can you try to run this with simulator to verify its working?

[falibabaei]

Hi,
Thank you for your reply.
I am using simulator.
With NVFlare version 2.4.0rc5 no I can not run the cifar_10. I have two errors. First, I encourage the error
mode = "PASSIVE"
root_path = ""
}': path: 'components.#2', exception: 'FileNotFoundError: [Errno 2] No such file or directory: '''

I inserted a space in root_path. Then I had an error:
2023-11-30 10:34:44,162 - PTClientAPILauncherExecutor - ERROR - [identity=site-1, run=simulate_job, peer=simulator_server, peer_run=simulate_job, task_name=train, task_id=4ddb604c-4495-4b44-8ab7-04a432470929]: missing result FLModel for train_task: train.

So I installed nvflare with pip install -e. This way it works. But my code does not work. I can not see any difference between my code and the cifar_10 example. It has the same structure.