src/server/worker/api_server.go
func (a *APIServer) worker() {
	logger := a.getWorkerLogger() // this worker's formatting logger

	// claim a shard if one is available or becomes available
	go a.claimShard(a.pachClient.Ctx())

	// Process incoming jobs
	backoff.RetryNotify(func() (retErr error) {
		retryCtx, retryCancel := context.WithCancel(a.pachClient.Ctx())
		defer retryCancel()
		watcher, err := a.jobs.ReadOnly(retryCtx).WatchByIndex(ppsdb.JobsPipelineIndex, a.pipelineInfo.Pipeline)
		if err != nil {
			return fmt.Errorf("error creating watch: %v", err)
		}
		defer watcher.Close()
	NextJob:
		for e := range watcher.Watch() {
			// Clear chunk caches from previous job
			if err := a.chunkCache.Clear(); err != nil {
				logger.Logf("error clearing chunk cache: %v", err)
			}
			if err := a.chunkStatsCache.Clear(); err != nil {
				logger.Logf("error clearing chunk stats cache: %v", err)
			}
			if e.Type == watch.EventError {
				return fmt.Errorf("worker watch error: %v", e.Err)
			} else if e.Type == watch.EventDelete {
				// Job was deleted, e.g. because input commit was deleted. This is
				// handled by cancelCtxIfJobFails goro, which was spawned when job was
				// created. Nothing to do here
				continue
			}

			// 'e' is a Put event -- new job
			var jobID string
			jobPtr := &pps.EtcdJobInfo{}
			if err := e.Unmarshal(&jobID, jobPtr); err != nil {
				return fmt.Errorf("error unmarshalling: %v", err)
			}
			if ppsutil.IsTerminal(jobPtr.State) {
				// previously-created job has finished, or job was finished during backoff
				// or in the 'watcher' queue
				logger.Logf("skipping job %v as it is already in state %v", jobID, jobPtr.State)
				continue NextJob
			}

			// create new ctx for this job, and don't use retryCtx as the
			// parent. Just because another job's etcd write failed doesn't
			// mean this job shouldn't run
			jobCtx, jobCancel := context.WithCancel(a.pachClient.Ctx())
			defer jobCancel() // cancel the job ctx
			pachClient := a.pachClient.WithCtx(jobCtx)

			//  Watch for any changes to EtcdJobInfo corresponding to jobID; if
			// the EtcdJobInfo is marked 'FAILED', call jobCancel().
			// ('watcher' above can't detect job state changes--it's watching
			// an index and so only emits when jobs are created or deleted).
			go a.cancelCtxIfJobFails(jobCtx, jobCancel, jobID)

			// Inspect the job and make sure it's relevant, as this worker may be old
			jobInfo, err := pachClient.InspectJob(jobID, false)
			if err != nil {
				if col.IsErrNotFound(err) {
					continue NextJob // job was deleted--no sense retrying
				}
				return fmt.Errorf("error from InspectJob(%v): %+v", jobID, err)
			}
			if jobInfo.PipelineVersion < a.pipelineInfo.Version {
				continue
			}
			if jobInfo.PipelineVersion > a.pipelineInfo.Version {
				return fmt.Errorf("job %s's version (%d) greater than pipeline's "+
					"version (%d), this should automatically resolve when the worker "+
					"is updated", jobID, jobInfo.PipelineVersion, a.pipelineInfo.Version)
			}

			// Read the chunks laid out by the master and create the datum factory
			plan := &Plan{}
			if err := a.plans.ReadOnly(jobCtx).GetBlock(jobInfo.Job.ID, plan); err != nil {
				return err
			}
			df, err := NewDatumFactory(pachClient, jobInfo.Input)
			if err != nil {
				return fmt.Errorf("error from NewDatumFactory: %v", err)
			}

			// Compute the datums to skip
			skip := make(map[string]struct{})
			var useParentHashTree bool
			parentCommitInfo, err := a.getParentCommitInfo(jobCtx, pachClient, jobInfo.OutputCommit)
			if err != nil {
				return err
			}
			if parentCommitInfo != nil {
				var err error
				skip, err = a.getCommitDatums(jobCtx, pachClient, parentCommitInfo)
				if err != nil {
					return err
				}
				var count int
				for i := 0; i < df.Len(); i++ {
					files := df.Datum(i)
					datumHash := HashDatum(a.pipelineInfo.Pipeline.Name, a.pipelineInfo.Salt, files)
					if _, ok := skip[datumHash]; ok {
						count++
					}
				}
				if len(skip) == count {
					useParentHashTree = true
				}
			}

			// Get updated job info from master
			jobInfo, err = pachClient.InspectJob(jobID, false)
			if err != nil {
				return err
			}
			eg, ctx := errgroup.WithContext(jobCtx)
			// If a datum fails, acquireDatums updates the relevant lock in
			// etcd, which causes the master to fail the job (which is
			// handled above in the JOB_FAILURE case). There's no need to
			// handle failed datums here, just failed etcd writes.
			eg.Go(func() error {
				return a.acquireDatums(
					ctx, jobID, plan, logger,
					func(low, high int64) (*processResult, error) {
						processResult, err := a.processDatums(pachClient, logger, jobInfo, df, low, high, skip, useParentHashTree)
						if err != nil {
							return nil, err
						}
						return processResult, nil
					},
				)
			})
			eg.Go(func() error {
				return a.mergeDatums(ctx, pachClient, jobInfo, jobID, plan, logger, df, skip, useParentHashTree)
			})
			if err := eg.Wait(); err != nil {
				if jobCtx.Err() == context.Canceled {
					continue NextJob // job cancelled--don't restart, just wait for next job
				}
				return fmt.Errorf("acquire/process/merge datums for job %s exited with err: %v", jobID, err)
			}
		}
		return fmt.Errorf("worker: jobs.WatchByIndex(pipeline = %s) closed unexpectedly", a.pipelineInfo.Pipeline.Name)
	}, backoff.NewInfiniteBackOff(), func(err error, d time.Duration) error {
		logger.Logf("worker: watch closed or error running the worker process: %v; retrying in %v", err, d)
		return nil
	})
}
func (a *APIServer) worker() {
	logger := a.getWorkerLogger() // this worker's formatting logger

	// claim a shard if one is available or becomes available
	go a.claimShard(a.pachClient.Ctx())

	// Process incoming jobs
	backoff.RetryNotify(func() (retErr error) {
		retryCtx, retryCancel := context.WithCancel(a.pachClient.Ctx())
		defer retryCancel()
		watcher, err := a.jobs.ReadOnly(retryCtx).WatchByIndex(ppsdb.JobsPipelineIndex, a.pipelineInfo.Pipeline)
		if err != nil {
			return fmt.Errorf("error creating watch: %v", err)
		}
		defer watcher.Close()
	NextJob:
		for e := range watcher.Watch() {
			// Clear chunk caches from previous job
			if err := a.chunkCache.Clear(); err != nil {
				logger.Logf("error clearing chunk cache: %v", err)
			}
			if err := a.chunkStatsCache.Clear(); err != nil {
				logger.Logf("error clearing chunk stats cache: %v", err)
			}
			if e.Type == watch.EventError {
				return fmt.Errorf("worker watch error: %v", e.Err)
			} else if e.Type == watch.EventDelete {
				// Job was deleted, e.g. because input commit was deleted. This is
				// handled by cancelCtxIfJobFails goro, which was spawned when job was
				// created. Nothing to do here
				continue
			}

			// 'e' is a Put event -- new job
			var jobID string
			jobPtr := &pps.EtcdJobInfo{}
			if err := e.Unmarshal(&jobID, jobPtr); err != nil {
				return fmt.Errorf("error unmarshalling: %v", err)
			}
			if ppsutil.IsTerminal(jobPtr.State) {
				// previously-created job has finished, or job was finished during backoff
				// or in the 'watcher' queue
				logger.Logf("skipping job %v as it is already in state %v", jobID, jobPtr.State)
				continue NextJob
			}

			// create new ctx for this job, and don't use retryCtx as the
			// parent. Just because another job's etcd write failed doesn't
			// mean this job shouldn't run
			jobCtx, jobCancel := context.WithCancel(a.pachClient.Ctx())
			defer jobCancel() // cancel the job ctx
			pachClient := a.pachClient.WithCtx(jobCtx)

			//  Watch for any changes to EtcdJobInfo corresponding to jobID; if
			// the EtcdJobInfo is marked 'FAILED', call jobCancel().
			// ('watcher' above can't detect job state changes--it's watching
			// an index and so only emits when jobs are created or deleted).
			go a.cancelCtxIfJobFails(jobCtx, jobCancel, jobID)

			// Inspect the job and make sure it's relevant, as this worker may be old
			jobInfo, err := pachClient.InspectJob(jobID, false)
			if err != nil {
				if col.IsErrNotFound(err) {
					continue NextJob // job was deleted--no sense retrying
				}
				return fmt.Errorf("error from InspectJob(%v): %+v", jobID, err)
			}
			if jobInfo.PipelineVersion < a.pipelineInfo.Version {
				continue
			}
			if jobInfo.PipelineVersion > a.pipelineInfo.Version {
				return fmt.Errorf("job %s's version (%d) greater than pipeline's "+
					"version (%d), this should automatically resolve when the worker "+
					"is updated", jobID, jobInfo.PipelineVersion, a.pipelineInfo.Version)
			}

			// Read the chunks laid out by the master and create the datum factory
			plan := &Plan{}
			if err := a.plans.ReadOnly(jobCtx).GetBlock(jobInfo.Job.ID, plan); err != nil {
				return err
			}
			df, err := NewDatumFactory(pachClient, jobInfo.Input)
			if err != nil {
				return fmt.Errorf("error from NewDatumFactory: %v", err)
			}

			// Compute the datums to skip
			skip := make(map[string]struct{})
			var useParentHashTree bool
			parentCommitInfo, err := a.getParentCommitInfo(jobCtx, pachClient, jobInfo.OutputCommit)
			if err != nil {
				return err
			}
			if parentCommitInfo != nil {
				var err error
				skip, err = a.getCommitDatums(jobCtx, pachClient, parentCommitInfo)
				if err != nil {
					return err
				}
				var count int
				for i := 0; i < df.Len(); i++ {
					files := df.Datum(i)
					datumHash := HashDatum(a.pipelineInfo.Pipeline.Name, a.pipelineInfo.Salt, files)
					if _, ok := skip[datumHash]; ok {
						count++
					}
				}
				if len(skip) == count {
					useParentHashTree = true
				}
			}

			// Get updated job info from master
			jobInfo, err = pachClient.InspectJob(jobID, false)
			if err != nil {
				return err
			}
			eg, ctx := errgroup.WithContext(jobCtx)
			// If a datum fails, acquireDatums updates the relevant lock in
			// etcd, which causes the master to fail the job (which is
			// handled above in the JOB_FAILURE case). There's no need to
			// handle failed datums here, just failed etcd writes.
			eg.Go(func() error {
				return a.acquireDatums(
					ctx, jobID, plan, logger,
					func(low, high int64) (*processResult, error) {
						processResult, err := a.processDatums(pachClient, logger, jobInfo, df, low, high, skip, useParentHashTree)
						if err != nil {
							return nil, err
						}
						return processResult, nil
					},
				)
			})
			eg.Go(func() error {
				return a.mergeDatums(ctx, pachClient, jobInfo, jobID, plan, logger, df, skip, useParentHashTree)
			})
			if err := eg.Wait(); err != nil {
				if jobCtx.Err() == context.Canceled {
					continue NextJob // job cancelled--don't restart, just wait for next job
				}
				return fmt.Errorf("acquire/process/merge datums for job %s exited with err: %v", jobID, err)
			}
		}
		return fmt.Errorf("worker: jobs.WatchByIndex(pipeline = %s) closed unexpectedly", a.pipelineInfo.Pipeline.Name)
	}, backoff.NewInfiniteBackOff(), func(err error, d time.Duration) error {
		logger.Logf("worker: watch closed or error running the worker process: %v; retrying in %v", err, d)
		return nil
	})
}
// worker does the following:
//  - claims filesystem shards as they become available
//  - watches for new jobs (jobInfos in the jobs collection)
//  - claims chunks from the chunk layout it finds in the chunks collection
//  - claims those chunks with acquireDatums
//  - processes the chunks with processDatums
//  - merges the chunks with mergeDatums