func StartComputationPlugin(name string, compID StreamID) (ComputationPlugin, error) {
	log.Printf("[computations] Launching computation plugin '%s'", name)
	path := path.Join(os.Getenv("DAGGER_PLUGIN_PATH"), "computation-"+name)
	client, err := pie.StartProviderCodec(jsonrpc.NewClientCodec,
		os.Stderr,
		path,
	)
	if err != nil {
		return nil, fmt.Errorf("Error starting plugin %s: %s", name, err)
	}
	plugin := &computationPlugin{
		name:   name,
		compID: compID,
		client: client,
	}
	return plugin, nil
}
StartComputationPlugin starts the plugin process
package dagger
import (
	"encoding/json"
	"errors"
	"fmt"
	log "github.com/Sirupsen/logrus"
	"net"
	"net/rpc"
	"net/rpc/jsonrpc"
	"os"
	"path"
	"sync"
	"time"
	"github.com/natefinch/pie"
)
type statefulComputation struct {
	streamID     StreamID
	plugin       ComputationPlugin
	groupHandler GroupHandler
	linearizer   *Linearizer
	persister    TaskPersister
	lwmTracker   LWMTracker
	dispatcher   *StreamDispatcher
	stopCh       chan struct{}
	sync.RWMutex  
	initialized  bool
}
func newStatefulComputation(streamID StreamID, coordinator Coordinator,
	persister Persister, plugin ComputationPlugin, dispatcherConfig func(*DispatcherConfig)) (*statefulComputation, error) {
	groupHandler, err := coordinator.JoinGroup(streamID)
	if err != nil {
		return nil, err
	}
	stopCh := make(chan struct{})
	lwmTracker := NewLWMTracker()
	linearizer := NewLinearizer(streamID, persister, lwmTracker)
	dispatcher := NewStreamDispatcher(streamID, coordinator, persister, lwmTracker, groupHandler, dispatcherConfig)
	computation := &statefulComputation{
		streamID:     streamID,
		plugin:       plugin,
		groupHandler: groupHandler,
		persister:    persister,
		lwmTracker:   lwmTracker,
		linearizer:   linearizer,
		dispatcher:   dispatcher,
		stopCh:       stopCh,
		RWMutex:      sync.RWMutex{},
		initialized:  false,
	}
	linearizer.SetProcessor(computation)
	return computation, nil
}
func (comp *statefulComputation) GetSnapshot() ([]byte, error) {
	log.Println("[computations] trying to acquire sync lock...")
	comp.Lock()
	log.Println("[computations] ... sync lock acquired!")
	defer comp.Unlock()
	snapshot := make(map[string][]byte)
	persisterSnapshot, err := comp.persister.GetSnapshot(comp.streamID)
	if err != nil {
		return nil, errors.New("stateful computation: " + err.Error())
	}
	pluginSnapshot, err := comp.plugin.GetSnapshot()
	if err != nil {
		return nil, errors.New("stateful computation: " + err.Error())
	}
	snapshot["persister"] = persisterSnapshot
	snapshot["plugin"] = pluginSnapshot
	return json.Marshal(snapshot)
}
func (comp *statefulComputation) Sync() (Timestamp, error) {
	comp.Lock()
	defer comp.Unlock()
	var from Timestamp
	for !comp.initialized {
		log.Printf("[computations] Computation %s not initialized, syncing with group",
			comp.streamID)
		areWeLeader, currentLeader, err := comp.groupHandler.GetStatus()
		if err != nil {
			return from, err
		}
		if currentLeader == "" {
			log.Println("[computations] Leader of ", comp.streamID, "not yet chosen, waiting")
			time.Sleep(time.Second)  
			continue
		}
		if !areWeLeader {
			if err != nil {
				return from, err
			}
			leaderHandler, err := newMasterHandler(currentLeader)
			if err != nil {
				return from, fmt.Errorf("[computations] Error creating master handler: %s", err)
			}
			snapshot, err := leaderHandler.Sync(comp.streamID)
			if err != nil {
				return from, fmt.Errorf("[computations] Error syncing computation with master: %s", err)
			}
			snapshotMap := make(map[string][]byte)
			err = json.Unmarshal(snapshot, &snapshotMap)
			if err != nil {
				return from, fmt.Errorf("cannot unmarshal snapshot: %s:", err.Error())
			}
			err = comp.plugin.ApplySnapshot(snapshotMap["plugin"])
			if err != nil {
				return from, fmt.Errorf("[computations] Error setting computation plugin state: %s", err)
			}
			err = comp.persister.ApplySnapshot(comp.streamID, snapshotMap["persister"])
			if err != nil {
				return from, fmt.Errorf("[computations] Error applying computation snapshot: %s", err)
			}
			from, err = comp.persister.GetLastTimestamp(comp.streamID)
			if err != nil {
				from = Timestamp(0)
			}
			from++
			comp.linearizer.SetStartLWM(from)
		}
		comp.initialized = true
	}
	return from, nil
}
func (comp *statefulComputation) Run(errc chan error) {
	go comp.dispatcher.Run(errc)
	go comp.linearizer.Run(errc)
}
func (comp *statefulComputation) Stop() {
	comp.linearizer.Stop()
	comp.plugin.Stop()
	comp.dispatcher.Stop()
}
func (comp *statefulComputation) ProcessRecord(r *Record) error {
	log.Println("[linearizer] processing", r)
	err := comp.linearizer.ProcessRecord(r)
	if err != nil {
		return err
	}
	return nil
}
func (comp *statefulComputation) ProcessRecordLinearized(t *Record) error {
	comp.Lock()
	defer comp.Unlock()
	response, err := comp.plugin.SubmitRecord(t)
	if err != nil {
		log.Println("ERROR:", err)
		return err
	}
	err = comp.persister.CommitComputation(comp.streamID, t, response.Records)
	if err != nil {
		return err
	}
	areWeLeader, _, err := comp.groupHandler.GetStatus()
	if err != nil {
		return err
	}
	if areWeLeader {
		log.Println("WE ARE LEADER")
		comp.lwmTracker.BeforeDispatching(response.Records)
		ProcessMultipleRecords(comp.dispatcher, response.Records)
	} else {
		log.Println("WE ARE NOT LEADER", t)
	}
	return nil
}