num_workers > 0 gets stuck

[mlstoic]

When I use num_workers =0 for train_dataloader, val_dataloader, test_dataloader, the training finishes one epoch %100 quickly (although I get loss = NaN and I have not figure out what the issue is) with some warning that I should use larger num_workers and it suggests me to use num_workers = 16.

However, if I use num_workers > 0 it gets stuck at the validation sanity check and it does not go anywhere.

Can someone please shed some light on what the issue might be ? Thank you.

[yongen9696]

could you share part of your code?

[mlstoic]

My code is not working completely yet. For num_workers = 0, I get the result at the bottom with loss = NaN, and for num_workers > 0, it gets stuck without any output. So, I am copy-pasting the majority of the code in case the problem is somewhere not related to num_workers. Thank you.

train_size = int(len(features_df) * 0.8)
train_df, test_df = features_df.iloc[:train_size], features_df.iloc[train_size+1:]
train_df.shape, test_df.shape


## Scaling

scaler = MinMaxScaler(feature_range=(-1,1))
scaler = scaler.fit(train_df)

train_df = pd.DataFrame(
    scaler.transform(train_df),
    index=train_df.index,
    columns=train_df.columns
)

test_df = pd.DataFrame(
    scaler.transform(test_df),
    index=test_df.index,
    columns=test_df.columns
)


## Creating Sequences for LSTM

def create_sequences(input_data: pd.DataFrame, target_column, sequence_length):
    sequences = []
    data_size = len(input_data)
    for i in range (data_size - sequence_length):
        sequence = input_data[i: i+sequence_length]
        label_position = i + sequence_length
        label = input_data.iloc[label_position][target_column]
        sequences.append((sequence, label))
    return sequences
	

train_sequences = create_sequences(train_df, "change", p["SEQ_LEN"])
test_sequences = create_sequences(test_df, "change", p["SEQ_LEN"])

## Pytorch Dataset

class PrDataset(Dataset):

    def __init__(self, sequences):
        self.sequences = sequences

    def __len__(self):
        return len(self.sequences)
        
    def __getitem__(self, idx):
        sequence, label = self.sequences[idx]
        return dict(
            sequence = torch.Tensor(sequence.to_numpy()),
            label = torch.Tensor(np.asarray(label)).float()
        )
		

## Data Module Class

class PrDataModule(pl.LightningDataModule):
    
    def __init__(self, train_sequences, test_sequences, batch_size = p["BATCH_SIZE"]):
        super().__init__()
        self.train_sequences = train_sequences
        self.test_sequences = test_sequences
        self.batch_size = batch_size

    def setup(self):
        self.train_dataset = PrDataset(self.train_sequences)
        self.test_dataset = PrDataset(self.test_sequences)

    def train_dataloader(self):
        return DataLoader(
            self.train_dataset,
            batch_size= self.batch_size,
            shuffle= False,
            num_workers=0 # ORIGINAL num_workers=1
            )

    def val_dataloader(self):
        return DataLoader(
            self.test_dataset,
            batch_size= 1,
            shuffle= False,
            num_workers=0 # ORIGINAL num_workers=1
            )

    def test_dataloader(self):
        return DataLoader(
            self.test_dataset,
            batch_size= 1,
            shuffle= False,
            num_workers=0 # ORIGINAL num_workers=1
            )
			
			
class PrDataModule(pl.LightningDataModule):
    
    def __init__(self, train_sequences, test_sequences, batch_size = p["BATCH_SIZE"]):
        super().__init__()
        self.train_sequences = train_sequences
        self.test_sequences = test_sequences
        self.batch_size = batch_size

    def setup(self):
        self.train_dataset = PrDataset(self.train_sequences)
        self.test_dataset = PrDataset(self.test_sequences)

    def train_dataloader(self):
        return DataLoader(
            self.train_dataset,
            batch_size= self.batch_size,
            shuffle= False,
            num_workers=0 # ORIGINAL num_workers=0
            )

    def val_dataloader(self):
        return DataLoader(
            self.test_dataset,
            batch_size= 1,
            shuffle= False,
            num_workers=0 # ORIGINAL num_workers=1
            )

    def test_dataloader(self):
        return DataLoader(
            self.test_dataset,
            batch_size= 1,
            shuffle= False,
            num_workers=0 # ORIGINAL num_workers=1
            )
			
## Model Initialization

data_module = PrDataModule(train_sequences, test_sequences, batch_size= p["BATCH_SIZE"]  )
data_module.setup()


## Model 

class PrPredictionModel(nn.Module):


    def __init__(self, n_features, dropout, n_hidden, num_layers):
        super().__init__()
        self.n_hidden = n_hidden
        self.dropout = dropout
        self.num_layers = num_layers
        self.lstm = nn.LSTM(
            input_size=n_features,
            hidden_size=n_hidden,
            batch_first=True,
            num_layers=num_layers,
            dropout=dropout
            )

        self.regressor= nn.Linear(n_hidden,1)

    def forward(self,x):
        # self.lstm.flatten_parameters()# Original
        # _, (hidden, ) = self.lstm(x) # Original
        # out=hidden[-1]    # Original 

        lstm_out, _ = self.lstm(x)
        out = lstm_out[:,-1]

        return self.regressor(out)
	
	
## Lightning Module Original

class PrPredictor(pl.LightningModule):

    def __init__(self, n_features: int, learning_rate, dropout, n_hidden, num_layers):
        super().__init__() 
        self.learning_rate = learning_rate
        self.n_hidden = n_hidden
        self.dropout = dropout
        self.num_layers = num_layers
        self.model= PrPredictionModel(n_features, dropout, n_hidden, num_layers)
        self.criterion=nn.MSELoss()
        
    def forward(self, x, labels=None):
        output = self.model(x)
        loss=0
        if labels is not None:
            loss = self.criterion(output, labels.unsqueeze(dim=1))
        return loss, output
    
    def training_step(self, batch, batch_idx):
        sequences = batch["sequence"]
        labels =batch["label"]
        y_hat = self(sequences,labels)
        loss, outputs = self(sequences,labels)  
        self.log("train_loss", loss, prog_bar=True, logger=True)
        return  loss

    def validation_step(self, batch, batch_idx):
        sequences = batch["sequence"]
        labels =batch["label"]
        y_hat = self(sequences,labels)
        loss, outputs = self(sequences,labels) 
        self.log("val_loss", loss, prog_bar=True, logger=True)
        return loss

    def test_step(self, batch, batch_idx):
        sequences = batch["sequence"]
        labels =batch["label"]
        y_hat = self(sequences,labels)
        loss, outputs = self(sequences,labels) 
        self.log("test_loss", loss, prog_bar=True, logger=True)
        return loss

    def configure_optimizers(self):
        return optim.AdamW(self.parameters(), lr=self.learning_rate)


## Class Initialization

model = PrPredictor(n_features=train_df.shape[1], learning_rate=p["learning_rate"], n_hidden = p["n_hidden"],
        dropout = p["dropout"], num_layers = p["num_layers"])
		


## Checkpoint

checkpoint_callback = ModelCheckpoint(
dirpath= "checkpoints",
filename="best-checkpoi nt",
#save_top_k = 1,
verbose= True,
monitor= "val_loss",
mode = "min"
)

## Early Stopping
early_stopping_callback = EarlyStopping (
    monitor = "val_loss",
    patience= 2
)

## Trainer 
trainer = pl.Trainer(
	logger = logger,
	checkpoint_callback=checkpoint_callback,
	callbacks = [early_stopping_callback],
	max_epochs = p["N_EPOCHS"],
	gpus = 1,
	progress_bar_refresh_rate = 30,
	auto_lr_find=True 
	)

## Training 

trainer.fit(model,data_module)

##------------------------------------------------
Output with num_workers = 0 :

C:\Am\ML\Python\Project\pytorch\lib\site-packages\pytorch_lightning\core\datamodule.py:469: LightningDeprecationWarning: DataModule.setup has already been called, so it will not be called again. In v1.6 this behavior will change to always call DataModule.setup.
  rank_zero_deprecation(
LOCAL_RANK: 0 - CUDA_VISIBLE_DEVICES: [0]

  | Name      | Type                 | Params
---------------------------------------------------
0 | model     | PricePredictionModel | 202 K 
1 | criterion | MSELoss              | 0     
---------------------------------------------------
202 K     Trainable params
0         Non-trainable params
202 K     Total params
0.812     Total estimated model params size (MB)
Validation sanity check:   0%|          | 0/2 [00:00<?, ?it/s]C:\Am\ML\Python\Project\pytorch\lib\site-packages\pytorch_lightning\trainer\data_loading.py:116: UserWarning: The dataloader, val_dataloader 0, does not have many workers which may be a bottleneck. Consider increasing the value of the `num_workers` argument` (try 16 which is the number of cpus on this machine) in the `DataLoader` init to improve performance.
  rank_zero_warn(
Global seed set to 42
C:\Am\ML\Python\Project\pytorch\lib\site-packages\pytorch_lightning\trainer\data_loading.py:116: UserWarning: The dataloader, train_dataloader, does not have many workers which may be a bottleneck. Consider increasing the value of the `num_workers` argument` (try 16 which is the number of cpus on this machine) in the `DataLoader` init to improve performance.
  rank_zero_warn(
Epoch 0:  20%|█▉        | 2070/10552 [00:14<00:58, 144.48it/s, loss=nan, v_num=4, train_loss=nan.0]C:\Am\ML\Python\Project\pytorch\lib\site-packages\pytorch_lightning\trainer\trainer.py:686: UserWarning: Detected KeyboardInterrupt, attempting graceful shutdown...
  rank_zero_warn("Detected KeyboardInterrupt, attempting graceful shutdown...")
Epoch 0:  20%|█▉        | 2070/10552 [00:29<02:00, 70.40it/s, loss=nan, v_num=4, train_loss=nan.0]

[lyingflatDDD]

Hi @mlstoic , do you fix the problem! Can you help me, i think i have a problem the same with your .

[vanpersie32]

@mlstoic have you ever solved the problem? same problem here