func (w *worker) Start(frame sif.DataFrame) error {
	if frame == nil {
		return fmt.Errorf("dataFrame cannot be nil")
	}
	conn, err := w.mconnect()
	if err != nil {
		return err
	}
	defer conn.Close()
	lis, err := net.Listen("tcp", w.opts.connectionString())
	if err != nil {
		return fmt.Errorf("failed to listen: %v", err)
	}
	w.lifecycleLock.Lock()
	w.server = grpc.NewServer()
	w.lifecycleLock.Unlock()
	eframe, ok := frame.(itypes.ExecutableDataFrame)
	if !ok {
		return fmt.Errorf("dataFrame must be executable")
	}
	tmpDirPath := fmt.Sprintf("sif-worker-%d", w.opts.Port)
	tmpDir, err := ioutil.TempDir(w.opts.TempDir, tmpDirPath)
	if err != nil {
		return err
	}
	log.Printf("Using temporary directory: %s", tmpDirPath)
	defer os.RemoveAll(tmpDir)
	statsTracker := &istats.RunStatistics{}
	ctx := context.Background()
	defer ctx.Done()
	compressor := compress.NewLZ4Compressor()
	planExecutor := eframe.Optimize().Execute(ctx, &itypes.PlanExecutorConfig{
		NumWorkers:               w.opts.NumWorkers,
		TempFilePath:             tmpDir,
		CacheMemoryHighWatermark: w.opts.CacheMemoryHighWatermark,
		Streaming:                eframe.GetParent().GetDataSource().IsStreaming(),
		IgnoreRowErrors:          w.opts.IgnoreRowErrors,
		Compressor:               compressor,
	}, statsTracker, false)
	defer planExecutor.Stop()
	statsTracker.Start(planExecutor.GetNumStages())
	defer statsTracker.Finish()
	defer compressor.Destroy()
	pb.RegisterLifecycleServiceServer(w.server, createLifecycleServer(w))
	pb.RegisterExecutionServiceServer(w.server, createExecutionServer(w.logClient, planExecutor, statsTracker))
	pb.RegisterPartitionsServiceServer(w.server, createPartitionServer(planExecutor))
	stats.RegisterStatsSourceServiceServer(w.server, createStatsSource(statsTracker))
	ctx, cancel := context.WithTimeout(context.Background(), w.opts.RPCTimeout)
	defer cancel()
	var wg sync.WaitGroup
	wg.Add(1)
	asyncErrors := iutil.CreateAsyncErrorChannel()
	go w.asyncRegisterWithCoordinator(ctx, &wg, asyncErrors)
	if err = iutil.WaitAndFetchError(&wg, asyncErrors); err != nil {
		panic(err)  
	}
	log.Printf("[%s] ready to run job at %s", w.ID(), w.opts.connectionString())
	err = w.server.Serve(lis)
	if err != nil {
		return fmt.Errorf("failed to listen: %v", err)
	}
	return nil
}
Start the worker - will block the current thread
package cluster
import (
	"context"
	"fmt"
	"io/ioutil"
	"log"
	"net"
	"os"
	"sync"
	"time"
	"github.com/go-sif/sif"
	"github.com/go-sif/sif/internal/compress"
	pb "github.com/go-sif/sif/internal/rpc"
	istats "github.com/go-sif/sif/internal/stats"
	itypes "github.com/go-sif/sif/internal/types"
	iutil "github.com/go-sif/sif/internal/util"
	"github.com/go-sif/sif/stats"
	uuid "github.com/gofrs/uuid"
	"google.golang.org/grpc"
)
type worker struct {
	id            string
	opts          *NodeOptions
	server        *grpc.Server
	lifecycleLock sync.Mutex
	clusterClient pb.ClusterServiceClient
	logClient     pb.LogServiceClient
	jobFinishedWg sync.Mutex
}
func createWorker(opts *NodeOptions) (*worker, error) {
	ensureDefaultNodeOptionsValues(opts)
	id, err := uuid.NewV4()
	if err != nil {
		log.Fatalf("failed to generate UUID: %v", err)
	}
	res := &worker{id: id.String(), opts: opts}
	res.jobFinishedWg.Lock()
	return res, nil
}
func (w *worker) mconnect() (*grpc.ClientConn, error) {
	conn, err := grpc.Dial(w.opts.coordinatorConnectionString(), grpc.WithInsecure())
	if err != nil {
		return nil, fmt.Errorf("fail to dial: %v", err)
	}
	w.logClient = pb.NewLogServiceClient(conn)
	w.clusterClient = pb.NewClusterServiceClient(conn)
	return conn, nil
}
func (w *worker) register() error {
	log.Printf("[%s] attempting to register with coordinator at %s:%d", w.ID(), w.opts.CoordinatorHost, w.opts.CoordinatorPort)
	ctx, cancel := context.WithTimeout(context.Background(), w.opts.RPCTimeout)
	defer cancel()
	req := pb.MRegisterRequest{
		Id:   w.id,
		Port: int32(w.opts.Port),
	}
	if w.logClient == nil {
		log.Fatalf("[%s] Cannot register before dialing coordinator with mconnect()", w.ID())
	}
	_, err := w.clusterClient.RegisterWorker(ctx, &req)
	return err
}
func (w *worker) ID() string {
	return w.id
}
func (w *worker) IsCoordinator() bool {
	return false
}