func ExecuteLocalActivity(ctx Context, activity interface{}, args ...interface{}) Future {
	i := getWorkflowInterceptor(ctx)
	env := getWorkflowEnvironment(ctx)
	activityType := getActivityFunctionName(env.GetRegistry(), activity)
	ctx = WithValue(ctx, localActivityFnContextKey, activity)
	return i.ExecuteLocalActivity(ctx, activityType, args...)
}
ExecuteLocalActivity requests to run a local activity. A local activity is like a regular activity with some key
differences:
* Local activity is scheduled and run by the workflow worker locally.
* Local activity does not need Cadence server to schedule activity task and does not rely on activity worker.
* No need to register local activity.
* The parameter activity to ExecuteLocalActivity() must be a function.
* Local activity is for short living activities (usually finishes within seconds).
* Local activity cannot heartbeat.

Context can be used to pass the settings for this local activity.
For now there is only one setting for timeout to be set:
lao := LocalActivityOptions{
ScheduleToCloseTimeout: 5 * time.Second,
}
ctx := WithLocalActivityOptions(ctx, lao)
The timeout here should be relative shorter than the DecisionTaskStartToCloseTimeout of the workflow. If you need a
longer timeout, you probably should not use local activity and instead should use regular activity. Local activity is
designed to be used for short living activities (usually finishes within seconds).

Input args are the arguments that will to be passed to the local activity. The input args will be hand over directly
to local activity function without serialization/deserialization because we don't need to pass the input across process
boundary. However, the result will still go through serialization/deserialization because we need to record the result
as history to cadence server so if the workflow crashes, a different worker can replay the history without running
the local activity again.

If the activity failed to complete then the future get error would indicate the failure, and it can be one of
CustomError, TimeoutError, CanceledError, PanicError, GenericError.
You can cancel the pending activity by cancel the context(workflow.WithCancel(ctx)) and that will fail the activity
with error CanceledError.

ExecuteLocalActivity returns Future with local activity result or failure.
package internal
import (
	"errors"
	"fmt"
	"reflect"
	"strings"
	"time"
	"github.com/uber-go/tally"
	s "go.uber.org/cadence/.gen/go/shared"
	"go.uber.org/cadence/internal/common"
	"go.uber.org/cadence/internal/common/backoff"
	"go.uber.org/zap"
)
var (
	errDomainNotSet                  = errors.New("domain is not set")
	errTaskListNotSet                = errors.New("task list is not set")
	errWorkflowIDNotSet              = errors.New("workflowId is not set")
	errLocalActivityParamsBadRequest = errors.New("missing local activity parameters through context, check LocalActivityOptions")
	errActivityParamsBadRequest      = errors.New("missing activity parameters through context, check ActivityOptions")
	errWorkflowOptionBadRequest      = errors.New("missing workflow options through context, check WorkflowOptions")
	errSearchAttributesNotSet        = errors.New("search attributes is empty")
)
type (
	Channel interface {
		Receive(ctx Context, valuePtr interface{}) (ok bool)
		ReceiveAsync(valuePtr interface{}) (ok bool)
		ReceiveAsyncWithMoreFlag(valuePtr interface{}) (ok bool, more bool)
		Send(ctx Context, v interface{})
		SendAsync(v interface{}) (ok bool)
		Close()
	}
	Selector interface {
		AddReceive(c Channel, f func(c Channel, ok bool)) Selector
		AddSend(c Channel, v interface{}, f func()) Selector
		AddFuture(future Future, f func(f Future)) Selector
		AddDefault(f func())
		Select(ctx Context)
	}
	WaitGroup interface {
		Add(delta int)
		Done()
		Wait(ctx Context)
	}
	Future interface {
		Get(ctx Context, valuePtr interface{}) error
		IsReady() bool
	}
	Settable interface {
		Set(value interface{}, err error)
		SetValue(value interface{})
		SetError(err error)
		Chain(future Future)  
	}
	ChildWorkflowFuture interface {
		Future
		GetChildWorkflowExecution() Future
		SignalChildWorkflow(ctx Context, signalName string, data interface{}) Future
	}
	WorkflowType struct {
		Name string
	}
	WorkflowExecution struct {
		ID    string
		RunID string
	}
	EncodedValue struct {
		value         []byte
		dataConverter DataConverter
	}
	Version int
	ChildWorkflowOptions struct {
		Domain string
		WorkflowID string
		TaskList string
		ExecutionStartToCloseTimeout time.Duration
		TaskStartToCloseTimeout time.Duration
		WaitForCancellation bool
		WorkflowIDReusePolicy WorkflowIDReusePolicy
		RetryPolicy *RetryPolicy
		CronSchedule string
		Memo map[string]interface{}
		SearchAttributes map[string]interface{}
		ParentClosePolicy ParentClosePolicy
	}
)
type RegisterWorkflowOptions struct {
	Name string
	EnableShortName               bool
	DisableAlreadyRegisteredCheck bool
}
func RegisterWorkflow(workflowFunc interface{}) {
	RegisterWorkflowWithOptions(workflowFunc, RegisterWorkflowOptions{})
}
func RegisterWorkflowWithOptions(workflowFunc interface{}, opts RegisterWorkflowOptions) {
	registry := getGlobalRegistry()
	registry.RegisterWorkflowWithOptions(workflowFunc, opts)
}
func Await(ctx Context, condition func() bool) error {
	state := getState(ctx)
	defer state.unblocked()
	for !condition() {
		doneCh := ctx.Done()
		if doneCh != nil {
			if _, more := doneCh.ReceiveAsyncWithMoreFlag(nil); !more {
				return NewCanceledError("Await context cancelled")
			}
		}
		state.yield("Await")
	}
	return nil
}
func NewChannel(ctx Context) Channel {
	state := getState(ctx)
	state.dispatcher.channelSequence++
	return NewNamedChannel(ctx, fmt.Sprintf("chan-%v", state.dispatcher.channelSequence))
}
func NewNamedChannel(ctx Context, name string) Channel {
	env := getWorkflowEnvironment(ctx)
	return &channelImpl{name: name, dataConverter: getDataConverterFromWorkflowContext(ctx), env: env}
}
func NewBufferedChannel(ctx Context, size int) Channel {
	env := getWorkflowEnvironment(ctx)
	return &channelImpl{size: size, dataConverter: getDataConverterFromWorkflowContext(ctx), env: env}
}
func NewNamedBufferedChannel(ctx Context, name string, size int) Channel {
	env := getWorkflowEnvironment(ctx)
	return &channelImpl{name: name, size: size, dataConverter: getDataConverterFromWorkflowContext(ctx), env: env}
}
func NewSelector(ctx Context) Selector {
	state := getState(ctx)
	state.dispatcher.selectorSequence++
	return NewNamedSelector(ctx, fmt.Sprintf("selector-%v", state.dispatcher.selectorSequence))
}
func NewNamedSelector(ctx Context, name string) Selector {
	return &selectorImpl{name: name}
}
func NewWaitGroup(ctx Context) WaitGroup {
	f, s := NewFuture(ctx)
	return &waitGroupImpl{future: f, settable: s}
}
func Go(ctx Context, f func(ctx Context)) {
	state := getState(ctx)
	state.dispatcher.newCoroutine(ctx, f)
}
func GoNamed(ctx Context, name string, f func(ctx Context)) {
	state := getState(ctx)
	state.dispatcher.newNamedCoroutine(ctx, name, f)
}
func NewFuture(ctx Context) (Future, Settable) {
	impl := &futureImpl{channel: NewChannel(ctx).(*channelImpl)}
	return impl, impl
}
func (wc *workflowEnvironmentInterceptor) ExecuteWorkflow(ctx Context, workflowType string, inputArgs ...interface{}) (results []interface{}) {
	args := []reflect.Value{reflect.ValueOf(ctx)}
	for _, arg := range inputArgs {
		switch arg.(type) {
		case []byte:
			args = append(args, reflect.ValueOf(arg))
		default:
			args = append(args, reflect.ValueOf(arg).Elem())
		}
	}
	fnValue := reflect.ValueOf(wc.fn)
	retValues := fnValue.Call(args)
	for _, r := range retValues {
		results = append(results, r.Interface())
	}
	return
}
func ExecuteActivity(ctx Context, activity interface{}, args ...interface{}) Future {
	i := getWorkflowInterceptor(ctx)
	registry := getRegistryFromWorkflowContext(ctx)
	activityType := getActivityFunctionName(registry, activity)
	return i.ExecuteActivity(ctx, activityType, args...)
}
func (wc *workflowEnvironmentInterceptor) ExecuteActivity(ctx Context, typeName string, args ...interface{}) Future {
	dataConverter := getDataConverterFromWorkflowContext(ctx)
	registry := getRegistryFromWorkflowContext(ctx)
	future, settable := newDecodeFuture(ctx, typeName)
	activityType, err := getValidatedActivityFunction(typeName, args, registry)
	if err != nil {
		settable.Set(nil, err)
		return future
	}
	options := getActivityOptions(ctx)
	options, err = getValidatedActivityOptions(ctx)
	if err != nil {
		settable.Set(nil, err)
		return future
	}
	if sessionInfo := getSessionInfo(ctx); sessionInfo != nil {
		isCreationActivity := isSessionCreationActivity(typeName)
		if sessionInfo.sessionState == sessionStateFailed && !isCreationActivity {
			settable.Set(nil, ErrSessionFailed)
			return future
		}
		if sessionInfo.sessionState == sessionStateOpen && !isCreationActivity {
			oldTaskListName := options.TaskListName
			options.TaskListName = sessionInfo.tasklist
			defer func() {
				options.TaskListName = oldTaskListName
			}()
		}
	}
	header := getHeadersFromContext(ctx)
	input, err := encodeArgs(dataConverter, args)
	if err != nil {
		panic(err)
	}
	params := executeActivityParams{
		activityOptions: *options,
		ActivityType:    *activityType,
		Input:           input,
		DataConverter:   dataConverter,
		Header:          header,
	}
	ctxDone, cancellable := ctx.Done().(*channelImpl)
	cancellationCallback := &receiveCallback{}
	a := getWorkflowEnvironment(ctx).ExecuteActivity(params, func(r []byte, e error) {
		settable.Set(r, e)
		if cancellable {
			ctxDone.removeReceiveCallback(cancellationCallback)
		}
	})
	if cancellable {
		cancellationCallback.fn = func(v interface{}, more bool) bool {
			if ctx.Err() == ErrCanceled {
				wc.env.RequestCancelActivity(a.activityID)
			}
			return false
		}
		_, ok, more := ctxDone.receiveAsyncImpl(cancellationCallback)
		if ok || !more {
			cancellationCallback.fn(nil, more)
		}
	}
	return future
}