func (driver *MesosSchedulerDriver) start() (mesos.Status, error) {
	select {
	case <-driver.started:
		return driver.status, errors.New("Unable to Start: driver has already been started once.")
	default:  
	}
	log.Infoln("Starting the scheduler driver...")
	if driver.status != mesos.Status_DRIVER_NOT_STARTED {
		return driver.status, fmt.Errorf("Unable to Start, expecting driver status %s, but is %s:", mesos.Status_DRIVER_NOT_STARTED, driver.status)
	}
	if err := driver.messenger.Start(); err != nil {
		log.Errorf("Scheduler failed to start the messenger: %v\n", err)
		return driver.status, err
	}
	pid := driver.messenger.UPID()
	driver.self = &pid
	driver.status = mesos.Status_DRIVER_RUNNING
	close(driver.started)
	log.Infof("Mesos scheduler driver started with PID=%v", driver.self)
	listener := detector.OnMasterChanged(func(m *mesos.MasterInfo) {
		driver.messenger.Route(context.TODO(), driver.self, &mesos.InternalMasterChangeDetected{
			Master: m,
		})
	})
	if driver.masterDetector != nil {
		log.V(1).Infof("starting master detector %T: %+v", driver.masterDetector, driver.masterDetector)
		driver.masterDetector.Detect(listener)
		log.V(2).Infoln("master detector started")
	}
	return driver.status, nil
}
start expected to be guarded by eventLock
package scheduler
import (
	"errors"
	"fmt"
	"math"
	"math/rand"
	"net"
	"os/user"
	"sync"
	"sync/atomic"
	"time"
	"github.com/gogo/protobuf/proto"
	log "github.com/golang/glog"
	"github.com/mesos/mesos-go/api/v0/auth"
	"github.com/mesos/mesos-go/api/v0/detector"
	mesos "github.com/mesos/mesos-go/api/v0/mesosproto"
	"github.com/mesos/mesos-go/api/v0/mesosproto/scheduler"
	util "github.com/mesos/mesos-go/api/v0/mesosutil"
	"github.com/mesos/mesos-go/api/v0/mesosutil/process"
	"github.com/mesos/mesos-go/api/v0/messenger"
	"github.com/mesos/mesos-go/api/v0/messenger/sessionid"
	"github.com/mesos/mesos-go/api/v0/upid"
	"github.com/pborman/uuid"
	"golang.org/x/net/context"
)
const (
	defaultAuthenticationTimeout = 30 * time.Second  
	registrationRetryIntervalMax = float64(1 * time.Minute)
	registrationBackoffFactor    = 2 * time.Second
)
var (
	ErrDisconnected           = errors.New("disconnected from mesos master")
	errAuthenticationCanceled = errors.New("authentication canceled")
)
type ErrDriverAborted struct {
	Reason string
}
func (err *ErrDriverAborted) Error() string {
	if err.Reason != "" {
		return err.Reason
	}
	return "driver-aborted"
}
type authenticationAttempt struct {
	done     chan struct{}
	doneOnce sync.Once
}
func (a *authenticationAttempt) cancel() {
	if a != nil {
		a.doneOnce.Do(func() { close(a.done) })
	}
}
func (a *authenticationAttempt) inProgress() bool {
	if a != nil {
		select {
		case <-a.done:
			return false
		default:
			return true
		}
	}
	return false
}
type DriverConfig struct {
	Scheduler        Scheduler
	Framework        *mesos.FrameworkInfo
	Master           string
	Credential       *mesos.Credential                      
	WithAuthContext  func(context.Context) context.Context  
	HostnameOverride string                                 
	BindingAddress   net.IP                                 
	BindingPort      uint16                                 
	PublishedAddress net.IP                                 
	NewMessenger     func() (messenger.Messenger, error)    
	NewDetector      func() (detector.Master, error)        
}
type MesosSchedulerDriver struct {
	masterPid       *upid.UPID
	frameworkInfo   *mesos.FrameworkInfo
	self            *upid.UPID
	stopCh          chan struct{}
	status          mesos.Status
	messenger       messenger.Messenger
	masterDetector  detector.Master
	connected       bool
	connection      uuid.UUID
	failoverTimeout float64
	failover        bool
	cache           *schedCache
	updates         map[string]*mesos.StatusUpdate  
	tasks           map[string]*mesos.TaskInfo      
	credential      *mesos.Credential
	authenticated   bool
	authenticating  *authenticationAttempt
	reauthenticate  bool
	withAuthContext func(context.Context) context.Context
	dispatch        func(context.Context, *upid.UPID, proto.Message) error  
	started         chan struct{}                                           
	eventLock       sync.RWMutex                                            
	withScheduler   func(f func(s Scheduler))                               
	done            chan struct{}                                           
}
func NewMesosSchedulerDriver(config DriverConfig) (initializedDriver *MesosSchedulerDriver, err error) {
	if config.Scheduler == nil {
		err = fmt.Errorf("Scheduler callbacks required.")
	} else if config.Master == "" {
		err = fmt.Errorf("Missing master location URL.")
	} else if config.Framework == nil {
		err = fmt.Errorf("FrameworkInfo must be provided.")
	} else if config.Credential != nil && config.WithAuthContext == nil {
		err = fmt.Errorf("WithAuthContext must be provided when Credential != nil")
	}
	if err != nil {
		return
	}
	framework := proto.Clone(config.Framework).(*mesos.FrameworkInfo)
	if framework.GetUser() == "" {
		user, err := user.Current()
		if err != nil || user == nil {
			if err != nil {
				log.Warningf("Failed to obtain username: %v\n", err)
			} else {
				log.Warningln("Failed to obtain username.")
			}
			framework.User = proto.String("")
		} else {
			framework.User = proto.String(user.Username)
		}
	}
	hostname := util.GetHostname(config.HostnameOverride)
	if framework.GetHostname() == "" {
		framework.Hostname = proto.String(hostname)
	}
	driver := &MesosSchedulerDriver{
		frameworkInfo:   framework,
		stopCh:          make(chan struct{}),
		status:          mesos.Status_DRIVER_NOT_STARTED,
		cache:           newSchedCache(),
		credential:      config.Credential,
		failover:        framework.Id != nil && len(framework.Id.GetValue()) > 0,
		withAuthContext: config.WithAuthContext,
		started:         make(chan struct{}),
		done:            make(chan struct{}),
	}
	driver.withScheduler = driver.makeWithScheduler(config.Scheduler)
	if framework.FailoverTimeout != nil && *framework.FailoverTimeout > 0 {
		driver.failoverTimeout = *framework.FailoverTimeout * float64(time.Second)
		log.V(1).Infof("found failover_timeout = %v", time.Duration(driver.failoverTimeout))
	}
	newDetector := config.NewDetector
	if newDetector == nil {
		newDetector = func() (detector.Master, error) {
			return detector.New(config.Master)
		}
	}
	newMessenger := config.NewMessenger
	if newMessenger == nil {
		newMessenger = func() (messenger.Messenger, error) {
			process := process.New("scheduler")
			return messenger.ForHostname(process, hostname, config.BindingAddress, config.BindingPort, config.PublishedAddress)
		}
	}
	if driver.masterDetector, err = newDetector(); err != nil {
		return
	} else if driver.messenger, err = newMessenger(); err != nil {
		return
	} else if err = driver.init(); err != nil {
		return
	} else {
		initializedDriver = driver
	}
	return
}
func (driver *MesosSchedulerDriver) makeWithScheduler(cs Scheduler) func(func(Scheduler)) {
	schedQueue := make(chan func(s Scheduler))
	go func() {
		defer func() {
			close(driver.done)
			log.V(1).Infoln("finished processing scheduler events")
		}()
		for f := range schedQueue {
			f(cs)
		}
	}()
	var schedLock sync.Mutex  
	abort := int32(0)
	return func(f func(s Scheduler)) {
		const timeout = 1 * time.Second
		t := time.NewTimer(timeout)
		defer t.Stop()
		trySend := func() (done bool) {
			driver.eventLock.Unlock()
			schedLock.Lock()
			defer func() {
				schedLock.Unlock()
				driver.eventLock.Lock()
			}()
			if atomic.LoadInt32(&abort) == 1 {
				return true
			}
			select {
			case schedQueue <- f:
				done = true
			case <-driver.stopCh:
				done = true
			case <-t.C:
			}
			select {
			case <-driver.stopCh:
				if atomic.CompareAndSwapInt32(&abort, 0, 1) {
					defer close(schedQueue)
					log.V(1).Infoln("stopping scheduler event queue..")
					select {
					case schedQueue <- f:
					case <-t.C:
					}
				}
			default:
			}
			return
		}
		for !trySend() {
			t.Reset(timeout)  
		}
		if atomic.LoadInt32(&abort) == 1 {
			driver.withScheduler = func(f func(_ Scheduler)) {}
		}
	}
}
func (driver *MesosSchedulerDriver) context() context.Context {
	return sessionid.NewContext(context.TODO(), driver.connection.String())
}
func (driver *MesosSchedulerDriver) init() error {
	log.Infof("Initializing mesos scheduler driver\n")
	driver.dispatch = driver.messenger.Send
	type messageHandler func(context.Context, *upid.UPID, proto.Message)
	guarded := func(h messageHandler) messenger.MessageHandler {
		return messenger.MessageHandler(func(from *upid.UPID, msg proto.Message) {
			driver.eventLock.Lock()
			defer driver.eventLock.Unlock()
			h(driver.context(), from, msg)
		})
	}
	driver.messenger.Install(guarded(driver.frameworkRegistered), &mesos.FrameworkRegisteredMessage{})
	driver.messenger.Install(guarded(driver.frameworkReregistered), &mesos.FrameworkReregisteredMessage{})
	driver.messenger.Install(guarded(driver.resourcesOffered), &mesos.ResourceOffersMessage{})
	driver.messenger.Install(guarded(driver.resourceOfferRescinded), &mesos.RescindResourceOfferMessage{})
	driver.messenger.Install(guarded(driver.statusUpdated), &mesos.StatusUpdateMessage{})
	driver.messenger.Install(guarded(driver.slaveLost), &mesos.LostSlaveMessage{})
	driver.messenger.Install(guarded(driver.frameworkMessageRcvd), &mesos.ExecutorToFrameworkMessage{})
	driver.messenger.Install(guarded(driver.frameworkErrorRcvd), &mesos.FrameworkErrorMessage{})
	driver.messenger.Install(guarded(driver.exitedExecutor), &mesos.ExitedExecutorMessage{})
	driver.messenger.Install(guarded(driver.handleMasterChanged), &mesos.InternalMasterChangeDetected{})
	driver.messenger.Install(guarded(driver.handleAuthenticationResult), &mesos.InternalAuthenticationResult{})
	driver.messenger.Install(guarded(driver.handleNetworkError), &mesos.InternalNetworkError{})
	return nil
}
func (driver *MesosSchedulerDriver) handleNetworkError(_ context.Context, from *upid.UPID, pbMsg proto.Message) {
	msg := pbMsg.(*mesos.InternalNetworkError)
	if driver.status == mesos.Status_DRIVER_ABORTED {
		log.Info("ignoring network error because the driver is aborted.")
		return
	} else if !from.Equal(driver.self) {
		log.Errorf("ignoring network error because message received from upid '%v'", from)
		return
	} else if !driver.connected {
		log.V(1).Infof("ignoring network error since we're not currently connected")
		return
	}
	if driver.masterPid.String() == msg.GetPid() && driver.connection.String() == msg.GetSession() {
		log.V(3).Info("Disconnecting scheduler.")
		driver.masterPid = nil
		driver.connected = false
		driver.authenticated = false
		driver.withScheduler(func(s Scheduler) { s.Disconnected(driver) })
		log.Info("master disconnected")
	}
}
func (driver *MesosSchedulerDriver) handleMasterChanged(_ context.Context, from *upid.UPID, pbMsg proto.Message) {
	if driver.status == mesos.Status_DRIVER_ABORTED {
		log.Info("Ignoring master change because the driver is aborted.")
		return
	} else if !from.Equal(driver.self) {
		log.Errorf("ignoring master changed message received from upid '%v'", from)
		return
	}
	wasConnected := driver.connected
	driver.connected = false
	driver.authenticated = false
	alertScheduler := false
	if wasConnected {
		log.V(3).Info("Disconnecting scheduler.")
		driver.masterPid = nil
		alertScheduler = true
	}
	msg := pbMsg.(*mesos.InternalMasterChangeDetected)
	master := msg.Master
	if master != nil {
		log.Infof("New master %s detected\n", master.GetPid())
		pid, err := upid.Parse(master.GetPid())
		if err != nil {
			panic("Unable to parse Master's PID value.")  
		}
		driver.masterPid = pid  
		defer driver.tryAuthentication()
	} else {
		log.Infoln("No master detected.")
	}
	if alertScheduler {
		driver.withScheduler(func(s Scheduler) { s.Disconnected(driver) })
	}
}
func (driver *MesosSchedulerDriver) tryAuthentication() {
	if driver.authenticated {
		panic("already authenticated")
	}
	masterPid := driver.masterPid  
	if masterPid == nil {
		log.Info("skipping authentication attempt because we lost the master")
		return
	}
	if driver.authenticating.inProgress() {
		driver.authenticating.cancel()
		driver.reauthenticate = true
		return
	}
	if driver.credential != nil {
		authenticating := &authenticationAttempt{done: make(chan struct{})}
		go func() {
			defer authenticating.cancel()
			result := &mesos.InternalAuthenticationResult{
				Success:   proto.Bool(false),
				Completed: proto.Bool(false),
				Pid:       proto.String(masterPid.String()),
			}
			if err := driver.authenticate(masterPid, authenticating); err != nil {
				log.Errorf("Scheduler failed to authenticate: %v\n", err)
				if err == auth.AuthenticationFailed {
					result.Completed = proto.Bool(true)
				}
			} else {
				result.Completed = proto.Bool(true)
				result.Success = proto.Bool(true)
			}
			pid := driver.messenger.UPID()
			driver.messenger.Route(context.TODO(), &pid, result)
		}()
		driver.authenticating = authenticating
	} else {
		log.Infoln("No credentials were provided. " +
			"Attempting to register scheduler without authentication.")
		driver.authenticated = true
		go driver.doReliableRegistration(float64(registrationBackoffFactor))
	}
}
func (driver *MesosSchedulerDriver) handleAuthenticationResult(_ context.Context, from *upid.UPID, pbMsg proto.Message) {
	if driver.status != mesos.Status_DRIVER_RUNNING {
		log.V(1).Info("ignoring authenticate because driver is not running")
		return
	}
	if !from.Equal(driver.self) {
		log.Errorf("ignoring authentication result message received from upid '%v'", from)
		return
	}
	if driver.authenticated {
		panic("already authenticated")
	}
	if driver.masterPid == nil {
		log.Infoln("ignoring authentication result because master is lost")
		driver.authenticating.cancel()  
		driver.reauthenticate = false
		return
	}
	msg := pbMsg.(*mesos.InternalAuthenticationResult)
	if driver.reauthenticate || !msg.GetCompleted() || driver.masterPid.String() != msg.GetPid() {
		log.Infof("failed to authenticate with master %v: master changed", driver.masterPid)
		driver.authenticating.cancel()  
		driver.reauthenticate = false
		driver.tryAuthentication()
		return
	}
	if !msg.GetSuccess() {
		log.Errorf("master %v refused authentication", driver.masterPid)
		return
	}
	driver.authenticated = true
	go driver.doReliableRegistration(float64(registrationBackoffFactor))
}
func (driver *MesosSchedulerDriver) Status() mesos.Status {
	driver.eventLock.RLock()
	defer driver.eventLock.RUnlock()
	return driver.status
}
func (driver *MesosSchedulerDriver) Running() bool {
	driver.eventLock.RLock()
	defer driver.eventLock.RUnlock()
	return driver.status == mesos.Status_DRIVER_RUNNING
}
func (driver *MesosSchedulerDriver) Connected() bool {
	driver.eventLock.RLock()
	defer driver.eventLock.RUnlock()
	return driver.connected
}
func (driver *MesosSchedulerDriver) stopped() bool {
	return driver.status != mesos.Status_DRIVER_RUNNING
}
func (driver *MesosSchedulerDriver) frameworkRegistered(_ context.Context, from *upid.UPID, pbMsg proto.Message) {
	log.V(2).Infoln("Handling scheduler driver framework registered event.")
	msg := pbMsg.(*mesos.FrameworkRegisteredMessage)
	masterInfo := msg.GetMasterInfo()
	masterPid := masterInfo.GetPid()
	frameworkId := msg.GetFrameworkId()
	if driver.status == mesos.Status_DRIVER_ABORTED {
		log.Infof("ignoring FrameworkRegisteredMessage from master %s, driver is aborted", masterPid)
		return
	}
	if driver.connected {
		log.Infoln("ignoring FrameworkRegisteredMessage from master, driver is already connected", masterPid)
		return
	}
	if driver.stopped() {
		log.Infof("ignoring FrameworkRegisteredMessage from master %s, driver is stopped", masterPid)
		return
	}
	if !driver.masterPid.Equal(from) {
		log.Warningf("ignoring framework registered message because it was sent from '%v' instead of leading master '%v'", from, driver.masterPid)
		return
	}
	log.Infof("Framework registered with ID=%s\n", frameworkId.GetValue())
	driver.frameworkInfo.Id = frameworkId  
	driver.connected = true
	driver.failover = false
	driver.connection = uuid.NewUUID()
	driver.withScheduler(func(s Scheduler) { s.Registered(driver, frameworkId, masterInfo) })
}
func (driver *MesosSchedulerDriver) frameworkReregistered(_ context.Context, from *upid.UPID, pbMsg proto.Message) {
	log.V(1).Infoln("Handling Scheduler re-registered event.")
	msg := pbMsg.(*mesos.FrameworkReregisteredMessage)
	if driver.status == mesos.Status_DRIVER_ABORTED {
		log.Infoln("Ignoring FrameworkReregisteredMessage from master, driver is aborted!")
		return
	}
	if driver.connected {
		log.Infoln("Ignoring FrameworkReregisteredMessage from master,driver is already connected!")
		return
	}
	if !driver.masterPid.Equal(from) {
		log.Warningf("ignoring framework re-registered message because it was sent from '%v' instead of leading master '%v'", from, driver.masterPid)
		return
	}
	log.Infof("Framework re-registered with ID [%s] ", msg.GetFrameworkId().GetValue())
	driver.connected = true
	driver.failover = false
	driver.connection = uuid.NewUUID()
	driver.withScheduler(func(s Scheduler) { s.Reregistered(driver, msg.GetMasterInfo()) })
}
func (driver *MesosSchedulerDriver) resourcesOffered(_ context.Context, from *upid.UPID, pbMsg proto.Message) {
	log.V(2).Infoln("Handling resource offers.")
	msg := pbMsg.(*mesos.ResourceOffersMessage)
	if driver.status == mesos.Status_DRIVER_ABORTED {
		log.Infoln("Ignoring ResourceOffersMessage, the driver is aborted!")
		return
	}
	if !driver.connected {
		log.Infoln("Ignoring ResourceOffersMessage, the driver is not connected!")
		return
	}
	pidStrings := msg.GetPids()
	if len(pidStrings) != len(msg.Offers) {
		log.Errorln("Ignoring offers, Offer count does not match Slave PID count.")
		return
	}
	for i, offer := range msg.Offers {
		if pid, err := upid.Parse(pidStrings[i]); err == nil {
			driver.cache.putOffer(offer, pid)
			log.V(2).Infof("Cached offer %s from SlavePID %s", offer.Id.GetValue(), pid)
		} else {
			log.Warningf("Failed to parse offer PID '%v': %v", pid, err)
		}
	}
	driver.withScheduler(func(s Scheduler) { s.ResourceOffers(driver, msg.Offers) })
}
func (driver *MesosSchedulerDriver) resourceOfferRescinded(_ context.Context, from *upid.UPID, pbMsg proto.Message) {
	log.V(1).Infoln("Handling resource offer rescinded.")
	msg := pbMsg.(*mesos.RescindResourceOfferMessage)
	if driver.status == mesos.Status_DRIVER_ABORTED {
		log.Infoln("Ignoring RescindResourceOfferMessage, the driver is aborted!")
		return
	}
	if !driver.connected {
		log.Infoln("Ignoring ResourceOffersMessage, the driver is not connected!")
		return
	}
	log.V(1).Infoln("Rescinding offer ", msg.OfferId.GetValue())
	driver.cache.removeOffer(msg.OfferId)
	driver.withScheduler(func(s Scheduler) { s.OfferRescinded(driver, msg.OfferId) })
}
func (driver *MesosSchedulerDriver) send(ctx context.Context, upid *upid.UPID, msg proto.Message) error {
	c := make(chan error, 1)
	go func() { c <- driver.dispatch(ctx, upid, msg) }()
	select {
	case <-ctx.Done():
		<-c  
		return ctx.Err()
	case err := <-c:
		return err
	}
}
func (driver *MesosSchedulerDriver) statusUpdated(ctx context.Context, from *upid.UPID, pbMsg proto.Message) {
	msg := pbMsg.(*mesos.StatusUpdateMessage)
	if driver.status != mesos.Status_DRIVER_RUNNING {
		log.V(1).Infoln("Ignoring StatusUpdate message, the driver is not running!")
		return
	}
	if !from.Equal(driver.self) {
		if !driver.connected {
			log.V(1).Infoln("Ignoring StatusUpdate message, the driver is not connected!")
			return
		}
		if !driver.masterPid.Equal(from) {
			log.Warningf("ignoring status message because it was sent from '%v' instead of leading master '%v'", from, driver.masterPid)
			return
		}
	}
	log.V(2).Infof("Received status update from %q status source %q", from.String(), msg.GetPid())
	status := msg.Update.GetStatus()
	if clearUUID := len(msg.Update.Uuid) == 0 || from.Equal(driver.self) || msg.GetPid() == driver.self.String(); clearUUID {
		status.Uuid = nil
	} else {
		status.Uuid = msg.Update.Uuid
	}
	if driver.status == mesos.Status_DRIVER_ABORTED {
		log.V(1).Infoln("Not sending StatusUpdate ACK, the driver is aborted!")
	} else {
		ackRequired := len(msg.Update.Uuid) > 0 && !from.Equal(driver.self) && msg.GetPid() != driver.self.String()
		if ackRequired {
			ackMsg := &mesos.StatusUpdateAcknowledgementMessage{
				SlaveId:     msg.Update.SlaveId,
				FrameworkId: driver.frameworkInfo.Id,
				TaskId:      msg.Update.Status.TaskId,
				Uuid:        msg.Update.Uuid,
			}
			log.V(2).Infof("Sending ACK for status update %+v to %q", *msg.Update, from.String())
			if err := driver.send(ctx, driver.masterPid, ackMsg); err != nil {
				log.Errorf("Failed to send StatusUpdate ACK message: %v", err)
			}
		} else {
			log.V(2).Infof("Not sending ACK, update is not from slave %q", from.String())
		}
	}
	driver.withScheduler(func(s Scheduler) { s.StatusUpdate(driver, status) })
}
func (driver *MesosSchedulerDriver) exitedExecutor(_ context.Context, from *upid.UPID, pbMsg proto.Message) {
	log.V(1).Infoln("Handling ExitedExceutor event.")
	msg := pbMsg.(*mesos.ExitedExecutorMessage)
	if driver.status == mesos.Status_DRIVER_ABORTED {
		log.V(1).Infoln("Ignoring ExitedExecutor message, the driver is aborted!")
		return
	}
	if !driver.connected {
		log.V(1).Infoln("Ignoring ExitedExecutor message, the driver is not connected!")
		return
	}
	status := msg.GetStatus()
	log.V(2).Infof("Lost executor %q from slave %q for framework %q with status %d",
		msg.GetExecutorId().GetValue(),
		msg.GetSlaveId().GetValue(),
		msg.GetFrameworkId().GetValue(),
		status)
	driver.withScheduler(func(s Scheduler) { s.ExecutorLost(driver, msg.GetExecutorId(), msg.GetSlaveId(), int(status)) })
}
func (driver *MesosSchedulerDriver) slaveLost(_ context.Context, from *upid.UPID, pbMsg proto.Message) {
	log.V(1).Infoln("Handling LostSlave event.")
	msg := pbMsg.(*mesos.LostSlaveMessage)
	if driver.status == mesos.Status_DRIVER_ABORTED {
		log.V(1).Infoln("Ignoring LostSlave message, the driver is aborted!")
		return
	}
	if !driver.connected {
		log.V(1).Infoln("Ignoring LostSlave message, the driver is not connected!")
		return
	}
	log.V(2).Infoln("Lost slave ", msg.SlaveId.GetValue())
	driver.cache.removeSlavePid(msg.SlaveId)
	driver.withScheduler(func(s Scheduler) { s.SlaveLost(driver, msg.SlaveId) })
}
func (driver *MesosSchedulerDriver) frameworkMessageRcvd(_ context.Context, from *upid.UPID, pbMsg proto.Message) {
	log.V(1).Infoln("Handling framework message event.")
	msg := pbMsg.(*mesos.ExecutorToFrameworkMessage)
	if driver.status == mesos.Status_DRIVER_ABORTED {
		log.V(1).Infoln("Ignoring framwork message, the driver is aborted!")
		return
	}
	log.V(1).Infoln("Received Framwork Message ", msg.String())
	driver.withScheduler(func(s Scheduler) { s.FrameworkMessage(driver, msg.ExecutorId, msg.SlaveId, string(msg.Data)) })
}
func (driver *MesosSchedulerDriver) frameworkErrorRcvd(ctx context.Context, from *upid.UPID, pbMsg proto.Message) {
	log.V(1).Infoln("Handling framework error event.")
	msg := pbMsg.(*mesos.FrameworkErrorMessage)
	driver.fatal(ctx, msg.GetMessage())
}
func (driver *MesosSchedulerDriver) Start() (mesos.Status, error) {
	driver.eventLock.Lock()
	defer driver.eventLock.Unlock()
	return driver.start()
}