public synchronized void fit(@NonNull MultiDataSetIterator source) {
        stopFit.set(false);
        createZooIfNeccessary(true);

        if (!source.hasNext() && source.resetSupported())
            source.reset();

        MultiDataSetIterator iterator = source;
        if (prefetchSize > 0 && source.asyncSupported()) {
            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 AsyncMultiDataSetIterator(source, prefetchSize,
                                new LinkedBlockingQueue<>(prefetchSize * workers), true,
                                new InterleavedDataSetCallback(prefetchSize * 2));
            } else
                iterator = new AsyncMultiDataSetIterator(source, prefetchSize);
        }

        val locker = new AtomicInteger(0);

        val blockWrapper = new DummyBlockMultiDataSetIterator(iterator);

        var time1 = System.currentTimeMillis();
        while (blockWrapper.hasAnything() && !stopFit.get()) {
            if (modelParamsSupplier != null) {
                val params = modelParamsSupplier.get();
                if (params != null) {
                    if (zoo != null)
                        for (val z: zoo)
                            z.updateModelParams(params);
                }
            }

            if (updaterParamsSupplier != null) {
                val params = updaterParamsSupplier.get();
                if (params != null) {
                    if (zoo != null)
                        for (val z: zoo)
                            z.updateUpdaterParams(params);
                }
            }

            val dataSets = blockWrapper.next(workers);
            long time2 = System.currentTimeMillis();

            if (dataSets == null)
                throw new ND4JIllegalStateException("You can't have NULL as MultiDataSet");

            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);
            }

            /*
             now dataSet should be dispatched to next free workers, until all workers are busy. And then we should block till all finished.
            */

            for (int pos = 0; pos < dataSets.length; pos++) {
                zoo[pos].feedMultiDataSet(dataSets[pos], time2 - time1);
            }

            iterationsCounter.incrementAndGet();

            /*
             * if all workers are dispatched now, join till all are finished
             */
            for (int pos = 0; pos < dataSets.length; pos++) {
                zoo[pos].waitTillRunning();
            }

            //Nd4j.getMemoryManager().invokeGcOccasionally();

            // optional averaging
            if (zoo[0].averagingRequired() && iterationsCounter.get() % averagingFrequency == 0 ) {
                /*
                 * average model, and propagate it to all workers
                 */

                double score = getScore(locker);

                // averaging updaters state
                averageUpdatersState(locker, score);
            }

            locker.set(0);

            time1 = System.currentTimeMillis();
        }

        if (debug)
            log.info("Stopping everyone...");

        if (debug)
            log.info("Shutting down iterator...");

        if (prefetchSize > 0 && source.asyncSupported())
            ((AsyncMultiDataSetIterator) iterator).shutdown();

        /*
        // TODO: get rid of this code, 0 model is not replicated anyway
        // now we transfer models back from workers
        List<Model> models = new ArrayList<>();
        for (int i = 0; i < zoo.length; i++) {
            models.add(zoo[0].getModel());
        }
        
        // actual transfer code depends on trainer
        trainerContext.finalizeTraining(model, models.toArray(new Model[0]));
        */
        try {
            close();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

        // sanity checks, or the dataset may never average
        if (!wasAveraged)
            log.warn("Parameters were never averaged on current fit(). Ratios of batch size, num workers, and averaging frequency may be responsible.");
        //            throw new IllegalStateException("Parameters were never averaged. Please check batch size ratios, number of workers, and your averaging frequency.");

        log.debug("Iterations passed: {}", iterationsCounter.get());
        //        iterationsCounter.set(0);
    }
public synchronized void fit ( @ NonNull MultiDataSetIterator source ) { stopFit . set ( false ) ; createZooIfNeccessary ( true ) ; if ( ! source . hasNext ( ) && source . resetSupported ( ) ) source . reset ( ) ; MultiDataSetIterator iterator = source ; if ( prefetchSize > 0 && source . asyncSupported ( ) ) { 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 AsyncMultiDataSetIterator ( source , prefetchSize , new LinkedBlockingQueue < > ( prefetchSize * workers ) , true , new InterleavedDataSetCallback ( prefetchSize * 2 ) ) ; } else iterator = new AsyncMultiDataSetIterator ( source , prefetchSize ) ; } val locker = new AtomicInteger ( 0 ) ; val blockWrapper = new DummyBlockMultiDataSetIterator ( iterator ) ; var time1 = System . currentTimeMillis ( ) ; while ( blockWrapper . hasAnything ( ) && ! stopFit . get ( ) ) { if ( modelParamsSupplier != null ) { val params = modelParamsSupplier . get ( ) ; if ( params != null ) { if ( zoo != null ) for ( val z : zoo ) z . updateModelParams ( params ) ; } } if ( updaterParamsSupplier != null ) { val params = updaterParamsSupplier . get ( ) ; if ( params != null ) { if ( zoo != null ) for ( val z : zoo ) z . updateUpdaterParams ( params ) ; } } val dataSets = blockWrapper . next ( workers ) ; long time2 = System . currentTimeMillis ( ) ; if ( dataSets == null ) throw new ND4JIllegalStateException ( "You can't have NULL as MultiDataSet" ) ; locker . set ( dataSets . length ) ; if ( gradientsAccumulator != null && gradientsAccumulator instanceof Registerable ) { ( ( Registerable ) gradientsAccumulator ) . registerConsumers ( dataSets . length ) ; } for ( int pos = 0 ; pos < dataSets . length ; pos ++ ) { zoo [ pos ] . feedMultiDataSet ( dataSets [ pos ] , time2 - time1 ) ; } iterationsCounter . incrementAndGet ( ) ; for ( int pos = 0 ; pos < dataSets . length ; pos ++ ) { zoo [ pos ] . waitTillRunning ( ) ; } if ( zoo [ 0 ] . averagingRequired ( ) && iterationsCounter . get ( ) % averagingFrequency == 0 ) { double score = getScore ( locker ) ; averageUpdatersState ( locker , score ) ; } locker . set ( 0 ) ; time1 = System . currentTimeMillis ( ) ; } if ( debug ) log . info ( "Stopping everyone..." ) ; if ( debug ) log . info ( "Shutting down iterator..." ) ; if ( prefetchSize > 0 && source . asyncSupported ( ) ) ( ( AsyncMultiDataSetIterator ) iterator ) . shutdown ( ) ; try { close ( ) ; } catch ( Exception e ) { throw new RuntimeException ( e ) ; } if ( ! wasAveraged ) log . warn ( "Parameters were never averaged on current fit(). Ratios of batch size, num workers, and averaging frequency may be responsible." ) ; log . debug ( "Iterations passed: {}" , iterationsCounter . get ( ) ) ; }
@param source