You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by lc...@apache.org on 2018/01/17 17:57:38 UTC
[beam] branch go-sdk updated (864b359 -> c5a3ce0)
This is an automated email from the ASF dual-hosted git repository.
lcwik pushed a change to branch go-sdk
in repository https://gitbox.apache.org/repos/asf/beam.git.
from 864b359 BEAM-3474 Include stacks in panic messages.
new 0af2503 BEAM-3299: Add source reporting support.
new a86c25e Remove GetId() call from under lock.
new efa1c20 Add additional comments about concurrency invariants.
new c5a3ce0 Add initialization of active plans map.
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
sdks/go/pkg/beam/core/runtime/exec/datasource.go | 25 +++++
sdks/go/pkg/beam/core/runtime/exec/plan.go | 12 +++
sdks/go/pkg/beam/core/runtime/harness/harness.go | 117 +++++++++++++++++------
3 files changed, 125 insertions(+), 29 deletions(-)
--
To stop receiving notification emails like this one, please contact
['"commits@beam.apache.org" <co...@beam.apache.org>'].
[beam] 03/04: Add additional comments about concurrency invariants.
Posted by lc...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lcwik pushed a commit to branch go-sdk
in repository https://gitbox.apache.org/repos/asf/beam.git
commit efa1c20a9328177a6cd6ec4d744853417fd0ef48
Author: Bill Neubauer <wc...@google.com>
AuthorDate: Tue Jan 16 15:39:20 2018 -0800
Add additional comments about concurrency invariants.
Explain how gRPC-specific goroutines are managed.
Explain how plans are managed in the active and plans maps.
---
sdks/go/pkg/beam/core/runtime/harness/harness.go | 13 +++++++++++--
1 file changed, 11 insertions(+), 2 deletions(-)
diff --git a/sdks/go/pkg/beam/core/runtime/harness/harness.go b/sdks/go/pkg/beam/core/runtime/harness/harness.go
index 335e2b3..8822c62 100644
--- a/sdks/go/pkg/beam/core/runtime/harness/harness.go
+++ b/sdks/go/pkg/beam/core/runtime/harness/harness.go
@@ -90,6 +90,11 @@ func Main(ctx context.Context, loggingEndpoint, controlEndpoint string) error {
}
var cpuProfBuf bytes.Buffer
+
+ // gRPC requires all readers of a stream be the same goroutine, so this goroutine
+ // is responsible for managing the network data. All it does is pull data from
+ // the stream, and hand off the message to a goroutine to actually be handled,
+ // so as to avoid blocking the underlying network channel.
for {
req, err := client.Recv()
if err != nil {
@@ -132,10 +137,14 @@ func Main(ctx context.Context, loggingEndpoint, controlEndpoint string) error {
}
type control struct {
- plans map[string]*exec.Plan // protected by mu
+ // plans that are candidates for execution.
+ plans map[string]*exec.Plan // protected by mu
+ // plans that are actively being executed.
+ // a plan can only be in one of these maps at any time.
active map[string]*exec.Plan // protected by mu
mu sync.Mutex
- data *DataManager
+
+ data *DataManager
}
func (c *control) handleInstruction(ctx context.Context, req *fnpb.InstructionRequest) *fnpb.InstructionResponse {
--
To stop receiving notification emails like this one, please contact
"commits@beam.apache.org" <co...@beam.apache.org>.
[beam] 02/04: Remove GetId() call from under lock.
Posted by lc...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lcwik pushed a commit to branch go-sdk
in repository https://gitbox.apache.org/repos/asf/beam.git
commit a86c25e372c39e6a6780fa262ad45105a46e00b9
Author: Bill Neubauer <wc...@google.com>
AuthorDate: Tue Jan 16 14:29:08 2018 -0800
Remove GetId() call from under lock.
Don't want to have anything happening while holding the plan
lock that isn't obvious it can't fail.
---
sdks/go/pkg/beam/core/runtime/harness/harness.go | 6 ++++--
1 file changed, 4 insertions(+), 2 deletions(-)
diff --git a/sdks/go/pkg/beam/core/runtime/harness/harness.go b/sdks/go/pkg/beam/core/runtime/harness/harness.go
index 72db1f1..335e2b3 100644
--- a/sdks/go/pkg/beam/core/runtime/harness/harness.go
+++ b/sdks/go/pkg/beam/core/runtime/harness/harness.go
@@ -176,10 +176,12 @@ func (c *control) handleInstruction(ctx context.Context, req *fnpb.InstructionRe
if err != nil {
return fail(id, "translation failed: %v", err)
}
- log.Debugf(ctx, "Plan %v: %v", desc.GetId(), plan)
+
+ pid := desc.GetId()
+ log.Debugf(ctx, "Plan %v: %v", pid, plan)
c.mu.Lock()
- c.plans[desc.GetId()] = plan
+ c.plans[pid] = plan
c.mu.Unlock()
}
--
To stop receiving notification emails like this one, please contact
"commits@beam.apache.org" <co...@beam.apache.org>.
[beam] 01/04: BEAM-3299: Add source reporting support.
Posted by lc...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lcwik pushed a commit to branch go-sdk
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 0af2503d72c20dd88b9e63a557c63319068d70af
Author: Bill Neubauer <wc...@google.com>
AuthorDate: Sun Jan 14 20:15:13 2018 -0800
BEAM-3299: Add source reporting support.
Makes the harness sufficiently multithreaded so we can handle progress
reports while we are performing work. The plans maintaing their ability
to be reused, but still can't be used concurrently.
---
sdks/go/pkg/beam/core/runtime/exec/datasource.go | 25 ++++++
sdks/go/pkg/beam/core/runtime/exec/plan.go | 12 +++
sdks/go/pkg/beam/core/runtime/harness/harness.go | 97 ++++++++++++++++++------
3 files changed, 109 insertions(+), 25 deletions(-)
diff --git a/sdks/go/pkg/beam/core/runtime/exec/datasource.go b/sdks/go/pkg/beam/core/runtime/exec/datasource.go
index e29239b..6a87978 100644
--- a/sdks/go/pkg/beam/core/runtime/exec/datasource.go
+++ b/sdks/go/pkg/beam/core/runtime/exec/datasource.go
@@ -19,9 +19,12 @@ import (
"context"
"fmt"
"io"
+ "sync/atomic"
+ "time"
"github.com/apache/beam/sdks/go/pkg/beam/core/graph"
"github.com/apache/beam/sdks/go/pkg/beam/core/graph/coder"
+ "github.com/apache/beam/sdks/go/pkg/beam/log"
)
// DataSource is a Root execution unit.
@@ -32,6 +35,8 @@ type DataSource struct {
sid StreamID
source DataReader
+ count int64
+ start time.Time
}
func (n *DataSource) ID() UnitID {
@@ -45,6 +50,8 @@ func (n *DataSource) Up(ctx context.Context) error {
func (n *DataSource) StartBundle(ctx context.Context, id string, data DataManager) error {
n.sid = StreamID{Port: *n.Edge.Port, Target: *n.Edge.Target, InstID: id}
n.source = data
+ n.start = time.Now()
+ atomic.StoreInt64(&n.count, 0)
return n.Out.StartBundle(ctx, id, data)
}
@@ -93,6 +100,7 @@ func (n *DataSource) Process(ctx context.Context) error {
// Single chunk stream.
// log.Printf("Fixed size=%v", size)
+ atomic.AddInt64(&n.count, int64(size))
for i := int32(0); i < size; i++ {
value, err := cv.Decode(r)
@@ -116,6 +124,7 @@ func (n *DataSource) Process(ctx context.Context) error {
break
}
+ atomic.AddInt64(&n.count, int64(chunk))
for i := uint64(0); i < chunk; i++ {
value, err := cv.Decode(r)
if err != nil {
@@ -139,6 +148,7 @@ func (n *DataSource) Process(ctx context.Context) error {
ec := MakeElementDecoder(coder.SkipW(c))
for {
+ atomic.AddInt64(&n.count, 1)
t, err := DecodeWindowedValueHeader(c, r)
if err != nil {
if err == io.EOF {
@@ -163,6 +173,7 @@ func (n *DataSource) Process(ctx context.Context) error {
}
func (n *DataSource) FinishBundle(ctx context.Context) error {
+ log.Infof(context.Background(), "DataSource: %d elements in %d ns", n.count, time.Now().Sub(n.start))
n.sid = StreamID{}
n.source = nil
return n.Out.FinishBundle(ctx)
@@ -178,3 +189,17 @@ func (n *DataSource) String() string {
sid := StreamID{Port: *n.Edge.Port, Target: *n.Edge.Target}
return fmt.Sprintf("DataSource[%v] Out:%v", sid, n.Out.ID())
}
+
+// ProgressReportSnapshot captures the progress reading an input source.
+type ProgressReportSnapshot struct {
+ ID, Name string
+ Count int64
+}
+
+// Progress returns a snapshot of the source's progress.
+func (n *DataSource) Progress() ProgressReportSnapshot {
+ if n == nil {
+ return ProgressReportSnapshot{}
+ }
+ return ProgressReportSnapshot{n.sid.Target.ID, n.sid.Target.Name, atomic.LoadInt64(&n.count)}
+}
diff --git a/sdks/go/pkg/beam/core/runtime/exec/plan.go b/sdks/go/pkg/beam/core/runtime/exec/plan.go
index d26fb1f..732f315 100644
--- a/sdks/go/pkg/beam/core/runtime/exec/plan.go
+++ b/sdks/go/pkg/beam/core/runtime/exec/plan.go
@@ -32,11 +32,14 @@ type Plan struct {
units []Unit
status Status
+ source *DataSource
}
// NewPlan returns a new bundle execution plan from the given units.
func NewPlan(id string, units []Unit) (*Plan, error) {
var roots []Root
+ var source *DataSource
+
for _, u := range units {
if u == nil {
return nil, fmt.Errorf("no <nil> units")
@@ -44,6 +47,9 @@ func NewPlan(id string, units []Unit) (*Plan, error) {
if r, ok := u.(Root); ok {
roots = append(roots, r)
}
+ if s, ok := u.(*DataSource); ok {
+ source = s
+ }
}
if len(roots) == 0 {
return nil, fmt.Errorf("no root units")
@@ -54,6 +60,7 @@ func NewPlan(id string, units []Unit) (*Plan, error) {
status: Initializing,
roots: roots,
units: units,
+ source: source,
}, nil
}
@@ -137,3 +144,8 @@ func (p *Plan) String() string {
}
return fmt.Sprintf("Plan[%v]:\n%v", p.ID(), strings.Join(units, "\n"))
}
+
+// ProgressReport returns a snapshot of input progress of the plan.
+func (p *Plan) ProgressReport() ProgressReportSnapshot {
+ return p.source.Progress()
+}
diff --git a/sdks/go/pkg/beam/core/runtime/harness/harness.go b/sdks/go/pkg/beam/core/runtime/harness/harness.go
index 4ba3fb6..72db1f1 100644
--- a/sdks/go/pkg/beam/core/runtime/harness/harness.go
+++ b/sdks/go/pkg/beam/core/runtime/harness/harness.go
@@ -70,6 +70,9 @@ func Main(ctx context.Context, loggingEndpoint, controlEndpoint string) error {
respc := make(chan *fnpb.InstructionResponse, 100)
wg.Add(1)
+
+ // gRPC requires all writers to a stream be the same goroutine, so this is the
+ // goroutine for managing responses back to the control service.
go func() {
defer wg.Done()
for resp := range respc {
@@ -100,34 +103,39 @@ func Main(ctx context.Context, loggingEndpoint, controlEndpoint string) error {
return fmt.Errorf("recv failed: %v", err)
}
- log.Debugf(ctx, "RECV: %v", proto.MarshalTextString(req))
- recordInstructionRequest(req)
+ // Launch a goroutine to handle the control message.
+ // TODO(wcn): maybe only do this for 'heavy' messages?
+ // TODO(wcn): implement a rate limiter for 'heavy' messages?
+ go func() {
+ log.Debugf(ctx, "RECV: %v", proto.MarshalTextString(req))
+ recordInstructionRequest(req)
- if isEnabled("cpu_profiling") {
- cpuProfBuf.Reset()
- pprof.StartCPUProfile(&cpuProfBuf)
- }
- resp := ctrl.handleInstruction(ctx, req)
+ if isEnabled("cpu_profiling") {
+ cpuProfBuf.Reset()
+ pprof.StartCPUProfile(&cpuProfBuf)
+ }
+ resp := ctrl.handleInstruction(ctx, req)
- if isEnabled("cpu_profiling") {
- pprof.StopCPUProfile()
- if err := ioutil.WriteFile(fmt.Sprintf("%s/cpu_prof%s", storagePath, req.InstructionId), cpuProfBuf.Bytes(), 0644); err != nil {
- log.Warnf(ctx, "Failed to write CPU profile for instruction %s: %v", req.InstructionId, err)
+ if isEnabled("cpu_profiling") {
+ pprof.StopCPUProfile()
+ if err := ioutil.WriteFile(fmt.Sprintf("%s/cpu_prof%s", storagePath, req.InstructionId), cpuProfBuf.Bytes(), 0644); err != nil {
+ log.Warnf(ctx, "Failed to write CPU profile for instruction %s: %v", req.InstructionId, err)
+ }
}
- }
- recordInstructionResponse(resp)
- if resp != nil {
- respc <- resp
- }
+ recordInstructionResponse(resp)
+ if resp != nil {
+ respc <- resp
+ }
+ }()
}
}
type control struct {
- plans map[string]*exec.Plan
- data *DataManager
-
- // TODO: running pipelines
+ plans map[string]*exec.Plan // protected by mu
+ active map[string]*exec.Plan // protected by mu
+ mu sync.Mutex
+ data *DataManager
}
func (c *control) handleInstruction(ctx context.Context, req *fnpb.InstructionRequest) *fnpb.InstructionResponse {
@@ -140,7 +148,7 @@ func (c *control) handleInstruction(ctx context.Context, req *fnpb.InstructionRe
for _, desc := range msg.GetProcessBundleDescriptor() {
var roots []string
- for id, _ := range desc.GetTransforms() {
+ for id := range desc.GetTransforms() {
roots = append(roots, id)
}
p := &pb.Pipeline{
@@ -170,7 +178,9 @@ func (c *control) handleInstruction(ctx context.Context, req *fnpb.InstructionRe
}
log.Debugf(ctx, "Plan %v: %v", desc.GetId(), plan)
+ c.mu.Lock()
c.plans[desc.GetId()] = plan
+ c.mu.Unlock()
}
return &fnpb.InstructionResponse{
@@ -188,14 +198,27 @@ func (c *control) handleInstruction(ctx context.Context, req *fnpb.InstructionRe
log.Debugf(ctx, "PB: %v", msg)
ref := msg.GetProcessBundleDescriptorReference()
+ c.mu.Lock()
plan, ok := c.plans[ref]
+ // Make the plan active, and remove it from candidates
+ // since a plan can't be run concurrently.
+ c.active[id] = plan
+ delete(c.plans, ref)
+ c.mu.Unlock()
+
if !ok {
return fail(id, "execution plan for %v not found", ref)
}
- // TODO: Async execution. The below assumes serial bundle execution.
+ err := plan.Execute(ctx, id, c.data)
- if err := plan.Execute(ctx, id, c.data); err != nil {
+ // Move the plan back to the candidate state
+ c.mu.Lock()
+ c.plans[plan.ID()] = plan
+ delete(c.active, id)
+ c.mu.Unlock()
+
+ if err != nil {
return fail(id, "execute failed: %v", err)
}
@@ -209,12 +232,36 @@ func (c *control) handleInstruction(ctx context.Context, req *fnpb.InstructionRe
case req.GetProcessBundleProgress() != nil:
msg := req.GetProcessBundleProgress()
- log.Debugf(ctx, "PB Progress: %v", msg)
+ // log.Debugf(ctx, "PB Progress: %v", msg)
+
+ ref := msg.GetInstructionReference()
+ c.mu.Lock()
+ plan, ok := c.active[ref]
+ c.mu.Unlock()
+ if !ok {
+ return fail(id, "execution plan for %v not found", ref)
+ }
+
+ snapshot := plan.ProgressReport()
return &fnpb.InstructionResponse{
InstructionId: id,
Response: &fnpb.InstructionResponse_ProcessBundleProgress{
- ProcessBundleProgress: &fnpb.ProcessBundleProgressResponse{},
+ ProcessBundleProgress: &fnpb.ProcessBundleProgressResponse{
+ Metrics: &fnpb.Metrics{
+ Ptransforms: map[string]*fnpb.Metrics_PTransform{
+ snapshot.ID: &fnpb.Metrics_PTransform{
+ ProcessedElements: &fnpb.Metrics_PTransform_ProcessedElements{
+ Measured: &fnpb.Metrics_PTransform_Measured{
+ OutputElementCounts: map[string]int64{
+ snapshot.Name: snapshot.Count,
+ },
+ },
+ },
+ },
+ },
+ },
+ },
},
}
--
To stop receiving notification emails like this one, please contact
"commits@beam.apache.org" <co...@beam.apache.org>.
[beam] 04/04: Add initialization of active plans map.
Posted by lc...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lcwik pushed a commit to branch go-sdk
in repository https://gitbox.apache.org/repos/asf/beam.git
commit c5a3ce00eeb96ba4e357080d6bd27708273da295
Author: Bill Neubauer <wc...@google.com>
AuthorDate: Wed Jan 17 09:43:24 2018 -0800
Add initialization of active plans map.
---
sdks/go/pkg/beam/core/runtime/harness/harness.go | 5 +++--
1 file changed, 3 insertions(+), 2 deletions(-)
diff --git a/sdks/go/pkg/beam/core/runtime/harness/harness.go b/sdks/go/pkg/beam/core/runtime/harness/harness.go
index 8822c62..1dd0d2e 100644
--- a/sdks/go/pkg/beam/core/runtime/harness/harness.go
+++ b/sdks/go/pkg/beam/core/runtime/harness/harness.go
@@ -85,8 +85,9 @@ func Main(ctx context.Context, loggingEndpoint, controlEndpoint string) error {
}()
ctrl := &control{
- plans: make(map[string]*exec.Plan),
- data: &DataManager{},
+ plans: make(map[string]*exec.Plan),
+ active: make(map[string]*exec.Plan),
+ data: &DataManager{},
}
var cpuProfBuf bytes.Buffer
--
To stop receiving notification emails like this one, please contact
"commits@beam.apache.org" <co...@beam.apache.org>.