docs: public synchronized void fit(@NonNull DataSetIterator source) {
          log.info("Using workspaceMode {} for training", workspaceMode.name());
          stopFit.set(false);
          createZooIfNeccessary(false);
  
          if (!source.hasNext() && source.resetSupported())
              source.reset();
  
          DataSetIterator iterator = source;
  
          if (prefetchSize > 0 && source.asyncSupported()) {
              log.info("Creating asynchronous prefetcher...");
              if (isMQ) {
                  if (workers % Nd4j.getAffinityManager().getNumberOfDevices() != 0)
                      log.warn("Number of workers [{}] isn't optimal for available devices [{}]", workers,
                                      Nd4j.getAffinityManager().getNumberOfDevices());
  
                  iterator = new AsyncDataSetIterator(source, prefetchSize,
                                  new LinkedBlockingQueue<>(prefetchSize * workers), true,
                                  new InterleavedDataSetCallback(prefetchSize * 2));
  
              } else
                  iterator = new AsyncDataSetIterator(source, prefetchSize);
          }
  
  
          val nanos = new ArrayList<Long>();
          val locker = new AtomicInteger(0);
          var time1 = System.currentTimeMillis();
          log.info("Starting ParallelWrapper training round...");
          long intcnt = 0;
  
          val blockWrapper = new DummyBlockDataSetIterator(iterator);
  
          while (blockWrapper.hasAnything() && !stopFit.get()) {
              if (modelParamsSupplier != null) {
                  val params = modelParamsSupplier.get();
                  if (params != null) {
                      if (zoo != null) {
                          log.info("Updating model parameters...");
                          for (val z:zoo) {
                              z.updateModelParams(params);
                          }
                      }
                  }
              }
  
              if (updaterParamsSupplier != null) {
                  val params = updaterParamsSupplier.get();
                  if (params != null) {
                      if (zoo != null) {
                          log.info("Updating updater parameters...");
                          for (val z:zoo) {
                              z.updateUpdaterParams(params);
                          }
                      }
                  }
              }
  
              intcnt++;
              val dataSets = blockWrapper.next(workers);
              var time2 = System.currentTimeMillis();
              var lastEtlTime = time2 - time1;
  
              if (dataSets == null)
                  throw new ND4JIllegalStateException("You can't have NULL as DataSet");
  
              if (zoo == null)
                  throw new IllegalStateException(
                                  "ParallelWrapper.shutdown() has been called too early and will fail from this point forward.");
  
              locker.set(dataSets.length);
              /*
               * if we're using registerable accumulator (i.e. we're on spark or cuda with gradients sharing),
               * update it & notify about number of threads in this training round then
               */
              if (gradientsAccumulator != null && gradientsAccumulator instanceof Registerable) {
                  ((Registerable) gradientsAccumulator).registerConsumers(dataSets.length);
              }
  
  
              // feeding datasets
              for (int pos = 0; pos < dataSets.length; pos++) {
                  if (debug)
                      log.info("Feeding dataset {} to worker {}", intcnt, pos);
  
                  zoo[pos].feedDataSet(dataSets[pos], lastEtlTime);
              }
  
              iterationsCounter.incrementAndGet();
  
  
              // waiting till all threads are done
              for (int pos = 0; pos < dataSets.length; pos++) {
                  try {
                      zoo[pos].waitTillRunning();
                  } catch (Exception e) {
                      throw new RuntimeException(e);
                  }
              }
  
  
              // optional averaging
              if (iterationsCounter.get() % averagingFrequency == 0 && zoo[0].averagingRequired() ) {
                  long timeA1 = System.currentTimeMillis();
  
                  // model averaging happens within
                  double score = getScore(locker);
  
                  // updaters averging happens within (if any)
                  averageUpdatersState(locker, score);
  
                  long timeA2 = System.currentTimeMillis();
                  if (reportScore)
                      log.info("Averaging time: {} ms", timeA2 - timeA1);
              }
  
              time1 = System.currentTimeMillis();
              locker.set(0);
          }
  
          if (debug)
              log.info("Stopping everyone...");
  
          // ensure all threads stopped processing
          for (int cnt = 0; cnt < workers; cnt++) {
              try {
                  zoo[cnt].waitTillRunning();
              } catch (Exception e) {
                  throw new RuntimeException(e);
              }
          }
  
          if (debug)
              log.info("Shutting down iterator...");
  
          if (prefetchSize > 0 && source.asyncSupported())
              ((AsyncDataSetIterator) iterator).shutdown();
  
          try {
              close();
          } catch (Exception e) {
              throw new RuntimeException(e);
          }
  
          if (debug)
              log.info("Iterations passed: {}", iterationsCounter.get());
      }