forked from gocircuit/circuit
/
vexp.go
115 lines (99 loc) · 3.07 KB
/
vexp.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
// Copyright 2013 Tumblr, Inc.
// Use of this source code is governed by the license for
// The Go Circuit Project, found in the LICENSE file.
//
// Authors:
// 2013 Petar Maymounkov <p@gocircuit.org>
package lang
import (
//"fmt"
"log"
"reflect"
"sync"
"github.com/hoijui/circuit/pkg/use/n"
)
type exportGroup struct {
sync.Mutex
PtrPtr []*ptrPtrMsg
}
func (r *Runtime) exportValues(values []interface{}, importer n.Addr) ([]interface{}, []*ptrPtrMsg) {
eg := &exportGroup{}
rewriter := func(src, dst reflect.Value) bool {
return r.exportRewrite(src, dst, importer, eg)
}
// x := rewriteInterface(rewriter, values).([]interface{})
// return x, eg.PtrPtr
return rewriteInterface(rewriter, values).([]interface{}), eg.PtrPtr
}
func (r *Runtime) exportRewrite(src, dst reflect.Value, importer n.Addr, eg *exportGroup) bool {
// Serialize cross-runtime pointers
switch v := src.Interface().(type) {
case *_permptr:
pm := &permPtrPtrMsg{ID: v.impHandle().ID, TypeID: v.impHandle().Type.ID, Src: v.impHandle().Exporter}
dst.Set(reflect.ValueOf(pm))
return true
case *_ptr:
if importer == nil {
panic("exporting non-perm ptrptr without importer")
}
pm := &ptrPtrMsg{ID: v.impHandle().ID, Src: v.impHandle().Exporter}
dst.Set(reflect.ValueOf(pm))
eg.Lock()
eg.PtrPtr = append(eg.PtrPtr, pm)
eg.Unlock()
return true
case *_ref:
if importer == nil {
panic("exporting non-perm ptr without importer")
}
dst.Set(reflect.ValueOf(r.exportPtr(v.value, importer)))
return true
case *_permref:
dst.Set(reflect.ValueOf(r.exportPtr(v.value, nil)))
return true
}
return false
}
// exportPtr returns *permPtrMsg if importer is nil, and *ptrMsg otherwise.
func (r *Runtime) exportPtr(v interface{}, importer n.Addr) interface{} {
// Add exported value to export table
exph := r.exp.Add(v, importer)
// log.Printf("exporting %T with handle %s for importer %v", v, exph.ID.String(), importer)
if importer == nil {
return &permPtrMsg{ID: exph.ID, TypeID: exph.Type.ID}
}
// Monitor the importer for liveness.
// DropPtr the handles upon importer death.
r.lk.Lock()
defer r.lk.Unlock()
_, ok := r.live[importer.WorkerID()]
if !ok {
r.live[importer.WorkerID()] = struct{}{}
// The anonymous function creates a "lifeline" connection to the worker importing v.
// When this connections is broken, v is released.
go func() {
// Defer removal of v's handle from the export table to the end of this function
defer func() {
r.lk.Lock()
delete(r.live, importer.WorkerID())
r.lk.Unlock()
// DropPtr/forget all exported handles
r.exp.RemoveImporter(importer)
}()
conn, err := r.t.Dial(importer)
if err != nil {
// log.Println("problem dialing lifeline to", importer.String(), err.Error())
return
}
defer conn.Close()
if err = conn.Write(&dontReplyMsg{}); err != nil {
log.Println("problem writing on lifeline to", importer.String(), err.Error())
return
}
// Read returns when the remote dies and
// runs the conn into an error
conn.Read()
}()
}
return &ptrMsg{ID: exph.ID, TypeID: exph.Type.ID}
}