class_docstr: A task for training primal model and dual models jointly.
  It takes:
      - unlabelled source (aka source monolingual data for translation task),
      - unlabelled target (aka target monolingual data for translation task),
      - labelled (source, target) (aka parallel data for translation task),
class_docstr_tok: ["A","task","for","training","primal","model","and","dual","models","jointly",".","It","takes",":","-","unlabelled","source","(","aka","source","monolingual","data","for","translation","task",")",",","-","unlabelled","target","(","aka","target","monolingual","data","for","translation","task",")",",","-","labelled","(","source",",","target",")","(","aka","parallel","data","for","translation","task",")",","]
code: def train_step(
      self, sample, model, criterion, optimizer, ignore_grad=False
  ):
      # Apply unsupervised dual learning objectives to both types of
      # monolingual data
      self.num_update += 1
  
      model.train()
      agg_loss, agg_sample_size, agg_logging_output = 0.0, 0.0, {}
      for model_key in model.task_keys:
          data_keys = ["source", "parallel"]
          for data_key in data_keys:
              sample_key = f"{model_key}_{data_key}"
              if sample[sample_key] is None:
                  continue
              if data_key == "parallel":
                  loss, sample_size, logging_output = self.criterion(
                      model.models[model_key], sample[sample_key]
                  )
                  if ignore_grad:
                      loss *= 0
                  optimizer.backward(loss)
                  agg_loss += loss.detach().item()
                  agg_sample_size += sample_size
                  agg_logging_output[sample_key] = logging_output
              if (
                  data_key == "source"
                  and self.num_update > self.args.soft_updates
              ):
                  (
                      total_loss,
                      sample_size,
                      logging_output,
                  ) = self.task_criterion(
                      sample[sample_key],
                      model.models[model_key],
                      optimizer,
                      self._get_tgt_dict(model_key),
                      model.models[self._get_dual(model_key)],
                      optimizer,
                      self._get_src_dict(model_key),
                      len_penalty=self.args.length_penalty,
                      unk_reward=self.args.unk_reward,
                      word_reward=self.args.word_reward,
                  )
                  agg_loss += total_loss
                  agg_sample_size += sample_size
                  agg_logging_output[sample_key] = logging_output[
                      model_key
                  ]
  
      return agg_loss, agg_sample_size, agg_logging_output
  
imports: from collections import OrderedDict
  import torch
  from fairseq import optim, utils
  from fairseq.criterions import CRITERION_REGISTRY
  from fairseq.data import LanguagePairDataset, RoundRobinZipDatasets
  from fairseq.tasks import LegacyFairseqTask, register_task
  from pytorch_translate.data import dictionary, utils
  from pytorch_translate.dual_learning import dual_learning_models
  from pytorch_translate.dual_learning.dual_learning_criterion import UnsupervisedCriterion
  from pytorch_translate.tasks.pytorch_translate_task import PytorchTranslateTask
  from pytorch_translate.weighted_criterions import WeightedLabelSmoothedCrossEntropyCriterion
label_desc: Use image node_id 15 for calling the DualLearningTask obj's underlying member method code with example usage: obj.train_step(sample, model, criterion, optimizer, ignore_grad) and returns: agg_loss,  agg_sample_size,  agg_logging_output