diff --git a/.gitignore b/.gitignore index b57b01d..92268ea 100644 --- a/.gitignore +++ b/.gitignore @@ -1,5 +1,6 @@ # The experiments outputs out +out-*/ # The vendor dependencies vendor diff --git a/Makefile b/Makefile index 26ee5ff..70c6e18 100644 --- a/Makefile +++ b/Makefile @@ -19,6 +19,7 @@ LOG_LEVEL = info RETINA_CONTEXT_FILE = "./data/retina/es_hyper.neat.yml" RETINA_GENOME_FILE = "./data/retina/cppn_genome.yml" RETINA_TRIALS_COUNT = 1 +RETINA_MAX_WORKERS = 25 RETINA_OUT_DIR = $(OUT_DIR)/retina # The default targets to run @@ -35,6 +36,17 @@ execute-retina: --trials $(RETINA_TRIALS_COUNT) \ --log-level $(LOG_LEVEL) +# Run retina parallel experiment +# +execute-retina-parallel: + $(GORUN) executor.go --out $(RETINA_OUT_DIR) \ + --context $(RETINA_CONTEXT_FILE) \ + --genome $(RETINA_GENOME_FILE) \ + --experiment retina-parallel \ + --max-workers $(RETINA_MAX_WORKERS) \ + --trials $(RETINA_TRIALS_COUNT) \ + --log-level $(LOG_LEVEL) + # Run unit tests # test: @@ -56,4 +68,4 @@ $(OUT_DIR): clean: $(GOCLEAN) rm -f $(OUT_DIR)/$(BINARY_NAME) - rm -f $(OUT_DIR)/$(BINARY_UNIX) \ No newline at end of file + rm -f $(OUT_DIR)/$(BINARY_UNIX) diff --git a/examples/retina/retina-parallel.go b/examples/retina/retina-parallel.go new file mode 100644 index 0000000..c4aa9c3 --- /dev/null +++ b/examples/retina/retina-parallel.go @@ -0,0 +1,206 @@ +// Package retina provides implementation of the retina experiment +package retina + +import ( + "context" + "fmt" + "github.com/yaricom/goESHyperNEAT/v2/cppn" + "github.com/yaricom/goESHyperNEAT/v2/examples" + "github.com/yaricom/goNEAT/v4/experiment" + "github.com/yaricom/goNEAT/v4/experiment/utils" + "github.com/yaricom/goNEAT/v4/neat" + "github.com/yaricom/goNEAT/v4/neat/genetics" + "os" + "sync" + "time" +) + +type parallelGenerationEvaluator struct { + generationEvaluator + maxWorkers int +} + +type evaluationJob struct { + organism *genetics.Organism +} + +type evaluationJobResult struct { + genomeID int + fitness float64 + errorRate float64 + winner bool + solverLinks int + solverNodes int + err error +} + +func retinaEvaluationWorker(ctx context.Context, evaluator generationEvaluator, jobs <-chan evaluationJob, + results chan<- evaluationJobResult, wg *sync.WaitGroup) { + defer wg.Done() + + for job := range jobs { + winner, solver, err := evaluator.organismEvaluate(ctx, job.organism) + if err != nil { + results <- evaluationJobResult{err: err} + return + } + + results <- evaluationJobResult{ + genomeID: job.organism.Genotype.Id, + fitness: job.organism.Fitness, + errorRate: job.organism.Error, + winner: winner, + solverLinks: solver.LinkCount(), + solverNodes: solver.NodeCount(), + } + } +} + +// NewParallelGenerationEvaluator creates parallel generation evaluator for retina experiment. +// The numSpeciesTarget specifies the target number of species to maintain in the population. +// If the number of species differs from the numSpeciesTarget it will be automatically adjusted +// with compatAdjustFreq frequency, i.e., at each epoch % compatAdjustFreq == 0. +func NewParallelGenerationEvaluator(outDir string, env *Environment, numSpeciesTarget, compatAdjustFreq, + maxWorkers int) (experiment.GenerationEvaluator, experiment.TrialRunObserver) { + if maxWorkers < 1 { + maxWorkers = 1 + } + evaluator := ¶llelGenerationEvaluator{ + generationEvaluator: generationEvaluator{ + outDir: outDir, + env: env, + numSpeciesTarget: numSpeciesTarget, + compatAdjustFreq: compatAdjustFreq, + }, + maxWorkers: maxWorkers, + } + return evaluator, evaluator +} + +// GenerationEvaluate evaluates a population of organisms in parallel on the retina experiment. +func (e *parallelGenerationEvaluator) GenerationEvaluate(ctx context.Context, population *genetics.Population, + epoch *experiment.Generation) error { + options, ok := neat.FromContext(ctx) + if !ok { + return neat.ErrNEATOptionsNotFound + } + + popSize := len(population.Organisms) + resultsChan := make(chan evaluationJobResult, popSize) + jobsChan := make(chan evaluationJob, popSize) + organismMapping := make(map[int]*genetics.Organism, popSize) + + var wg sync.WaitGroup + for i := 0; i < e.maxWorkers; i++ { + wg.Add(1) + // Create private evaluator instance per worker to avoid races on mutable state. + workerEvaluator := generationEvaluator{ + outDir: e.outDir, + env: e.env, + numSpeciesTarget: e.numSpeciesTarget, + compatAdjustFreq: e.compatAdjustFreq, + } + go retinaEvaluationWorker(ctx, workerEvaluator, jobsChan, resultsChan, &wg) + } + + startTime := time.Now() + for _, organism := range population.Organisms { + if _, exists := organismMapping[organism.Genotype.Id]; exists { + return fmt.Errorf("organism with %d already exists in mapping", organism.Genotype.Id) + } + organismMapping[organism.Genotype.Id] = organism + jobsChan <- evaluationJob{organism: organism} + } + close(jobsChan) + + wg.Wait() + close(resultsChan) + + maxPopulationFitness := 0.0 + bestLinkCount := 0 + bestNodeCount := 0 + bestSolverLinks := -1 + bestSolverNodes := -1 + + for result := range resultsChan { + if result.err != nil { + return result.err + } + + organism, exists := organismMapping[result.genomeID] + if !exists { + return fmt.Errorf("organism not found in mapping for id: %d", result.genomeID) + } + + organism.Fitness = result.fitness + organism.Error = result.errorRate + + if organism.Fitness > maxPopulationFitness { + maxPopulationFitness = organism.Fitness + if phenotype, err := organism.Phenotype(); err == nil { + bestLinkCount = phenotype.LinkCount() + bestNodeCount = phenotype.NodeCount() + } else { + neat.ErrorLog(fmt.Sprintf("Failed to get organism Phenotype, reason: %s", err)) + return err + } + bestSolverLinks = result.solverLinks + bestSolverNodes = result.solverNodes + } + + if result.winner && (epoch.Champion == nil || organism.Fitness > epoch.Champion.Fitness) { + epoch.Solved = true + epoch.WinnerNodes = len(organism.Genotype.Nodes) + epoch.WinnerGenes = organism.Genotype.Extrons() + epoch.WinnerEvals = options.PopSize*epoch.Id + organism.Genotype.Id + epoch.Champion = organism + organism.IsWinner = true + } + } + elapsedTime := time.Since(startTime) + + epoch.FillPopulationStatistics(population) + + if epoch.Solved || epoch.Id%options.PrintEvery == 0 { + if _, err := utils.WritePopulationPlain(e.outDir, population, epoch); err != nil { + neat.ErrorLog(fmt.Sprintf("Failed to dump population, reason: %s\n", err)) + return err + } + } + + if epoch.Solved { + org := epoch.Champion + utils.PrintActivationDepth(org, true) + + genomeFile := "retina_cppn_winner" + if orgPath, err := utils.WriteGenomePlain(genomeFile, e.outDir, org, epoch); err != nil { + neat.ErrorLog(fmt.Sprintf("Failed to dump winner organism's genome, reason: %s\n", err)) + } else { + neat.InfoLog(fmt.Sprintf("Generation #%d winner's genome dumped to: %s\n", epoch.Id, orgPath)) + } + + graph := org.Data.Value.(cppn.SubstrateGraphBuilder) + nodes, _ := graph.NodesCount() + edges, _ := graph.EdgesCount() + substrPath := fmt.Sprintf("%s/%s_%d-%d.xml", utils.CreateOutDirForTrial(e.outDir, epoch.TrialId), + "retina_substrate_graph_winner", nodes, edges) + if file, err := os.Create(substrPath); err != nil { + neat.ErrorLog(err.Error()) + } else if err = graph.Marshal(file); err != nil { + neat.ErrorLog(fmt.Sprintf("Failed to dump winner substrate, reason: %s\n", err)) + } else { + neat.InfoLog(fmt.Sprintf("Generation #%d winner's substrate dumped to: %s\n", epoch.Id, substrPath)) + } + } else if epoch.Id < options.NumGenerations-1 { + speciesCount := len(population.Species) + + examples.AdjustSpeciesNumber(speciesCount, epoch.Id, e.compatAdjustFreq, e.numSpeciesTarget, options) + + neat.InfoLog( + fmt.Sprintf("%d species -> %d organisms [compatibility threshold: %.1f, target: %d]\nbest CPNN organism [fitness: %.2f, links: %d, nodes: %d], best solver [links: %d, nodes: %d], population evaluation time: %v", + speciesCount, len(population.Organisms), options.CompatThreshold, e.numSpeciesTarget, + maxPopulationFitness, bestLinkCount, bestNodeCount, bestSolverLinks, bestSolverNodes, elapsedTime)) + } + + return nil +} diff --git a/executor.go b/executor.go index 05c1e1a..10ce54f 100644 --- a/executor.go +++ b/executor.go @@ -21,7 +21,8 @@ func main() { var outDirPath = flag.String("out", "./out", "The output directory to store results.") var contextPath = flag.String("context", "./data/retina/es_hyper.neat.yml", "The execution context configuration file.") var genomePath = flag.String("genome", "./data/retina/cppn_genome.yml", "The seed genome to start with.") - var experimentName = flag.String("experiment", "retina", "The name of experiment to run. [retina]") + var experimentName = flag.String("experiment", "retina", "The name of experiment to run. [retina, retina-parallel]") + var maxWorkers = flag.Int("max-workers", 1, "Maximum number of workers to use for parallel experiments.") var speciesTarget = flag.Int("species_target", 15, "The target number of species to maintain.") var speciesCompatAdjustFreq = flag.Int("species_adjust_freq", 10, "The frequency of species compatibility threshold adjustments when trying to maintain their number.") @@ -82,6 +83,9 @@ func main() { if len(*logLevel) > 0 { neat.LogLevel = neat.LoggerLevel(*logLevel) } + if *maxWorkers < 1 { + *maxWorkers = 1 + } // Create Experiment experimentContext := neatOptions.NeatContext() @@ -106,6 +110,19 @@ func main() { generationEvaluator, trialObserver = retina.NewGenerationEvaluator( *outDirPath, env, *speciesTarget, *speciesCompatAdjustFreq) } + case "retina-parallel": + opts, err := eshyperneat.LoadYAMLConfigFile(*contextPath) + if err != nil { + log.Fatal("Failed to load ES-HyperNEAT options from config file: ", err) + } else { + experimentContext = eshyperneat.NewContext(experimentContext, opts) + } + if env, err := retina.NewRetinaEnvironment(retina.CreateRetinaDataset(), 4); err != nil { + log.Fatalf("Failed to create retina environment, reason: %s", err) + } else { + generationEvaluator, trialObserver = retina.NewParallelGenerationEvaluator( + *outDirPath, env, *speciesTarget, *speciesCompatAdjustFreq, *maxWorkers) + } default: log.Fatalf("Unsupported experiment name requested: %s\n", *experimentName) }