You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by ke...@apache.org on 2017/10/26 17:11:01 UTC
[4/6] beam git commit: [BEAM-3005] Set JVM max heap size in java
container
[BEAM-3005] Set JVM max heap size in java container
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/8370a8d9
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/8370a8d9
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/8370a8d9
Branch: refs/heads/master
Commit: 8370a8d91f2b86ddf2fefe5f755084a3973edc9f
Parents: 28fbb01
Author: Henning Rohde <he...@google.com>
Authored: Mon Oct 23 09:56:55 2017 -0700
Committer: Henning Rohde <he...@google.com>
Committed: Thu Oct 26 08:12:10 2017 -0700
----------------------------------------------------------------------
sdks/go/pkg/beam/artifact/server_test.go | 2 +-
sdks/go/pkg/beam/util/syscallx/syscall.go | 26 ++++++++++++++
.../pkg/beam/util/syscallx/syscall_default.go | 28 +++++++++++++++
sdks/go/pkg/beam/util/syscallx/syscall_linux.go | 38 ++++++++++++++++++++
sdks/java/container/boot.go | 23 +++++++++++-
5 files changed, 115 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/8370a8d9/sdks/go/pkg/beam/artifact/server_test.go
----------------------------------------------------------------------
diff --git a/sdks/go/pkg/beam/artifact/server_test.go b/sdks/go/pkg/beam/artifact/server_test.go
index bb1c06c..85f54a2 100644
--- a/sdks/go/pkg/beam/artifact/server_test.go
+++ b/sdks/go/pkg/beam/artifact/server_test.go
@@ -21,12 +21,12 @@ import (
"net"
"sync"
"testing"
+ "time"
pb "github.com/apache/beam/sdks/go/pkg/beam/model/jobmanagement_v1"
"github.com/apache/beam/sdks/go/pkg/beam/util/grpcx"
"golang.org/x/net/context"
"google.golang.org/grpc"
- "time"
)
// startServer starts an in-memory staging and retrieval artifact server
http://git-wip-us.apache.org/repos/asf/beam/blob/8370a8d9/sdks/go/pkg/beam/util/syscallx/syscall.go
----------------------------------------------------------------------
diff --git a/sdks/go/pkg/beam/util/syscallx/syscall.go b/sdks/go/pkg/beam/util/syscallx/syscall.go
new file mode 100644
index 0000000..151f860
--- /dev/null
+++ b/sdks/go/pkg/beam/util/syscallx/syscall.go
@@ -0,0 +1,26 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// Syscallx provide system call utilities that attempt to hide platform differences.
+// Operations returns UnsupportedErr if not implemented on the given platform, so
+// consumers of this package should generally treat that error specially.
+package syscallx
+
+import (
+ "errors"
+)
+
+// UnsupportedErr is the error returned for unsupported operations.
+var UnsupportedErr = errors.New("not supported on platform")
http://git-wip-us.apache.org/repos/asf/beam/blob/8370a8d9/sdks/go/pkg/beam/util/syscallx/syscall_default.go
----------------------------------------------------------------------
diff --git a/sdks/go/pkg/beam/util/syscallx/syscall_default.go b/sdks/go/pkg/beam/util/syscallx/syscall_default.go
new file mode 100644
index 0000000..ccc9324
--- /dev/null
+++ b/sdks/go/pkg/beam/util/syscallx/syscall_default.go
@@ -0,0 +1,28 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// +build !linux
+
+package syscallx
+
+// PhysicalMemorySize returns the total physical memory size.
+func PhysicalMemorySize() (uint64, error) {
+ return 0, UnsupportedErr
+}
+
+// FreeDiskSpace returns the free disk space for a given path.
+func FreeDiskSpace(path string) (uint64, error) {
+ return 0, UnsupportedErr
+}
http://git-wip-us.apache.org/repos/asf/beam/blob/8370a8d9/sdks/go/pkg/beam/util/syscallx/syscall_linux.go
----------------------------------------------------------------------
diff --git a/sdks/go/pkg/beam/util/syscallx/syscall_linux.go b/sdks/go/pkg/beam/util/syscallx/syscall_linux.go
new file mode 100644
index 0000000..c639f87
--- /dev/null
+++ b/sdks/go/pkg/beam/util/syscallx/syscall_linux.go
@@ -0,0 +1,38 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// +build linux
+
+package syscallx
+
+import "syscall"
+
+// PhysicalMemorySize returns the total physical memory size.
+func PhysicalMemorySize() (uint64, error) {
+ var info syscall.Sysinfo_t
+ if err := syscall.Sysinfo(&info); err != nil {
+ return 0, err
+ }
+ return info.Totalram, nil
+}
+
+// FreeDiskSpace returns the free disk space for a given path.
+func FreeDiskSpace(path string) (uint64, error) {
+ var stat syscall.Statfs_t
+ if err := syscall.Statfs(path, &stat); err != nil {
+ return 0, err
+ }
+ return stat.Bavail * uint64(stat.Bsize), nil
+}
http://git-wip-us.apache.org/repos/asf/beam/blob/8370a8d9/sdks/java/container/boot.go
----------------------------------------------------------------------
diff --git a/sdks/java/container/boot.go b/sdks/java/container/boot.go
index 2e140a1..a5505d5 100644
--- a/sdks/java/container/boot.go
+++ b/sdks/java/container/boot.go
@@ -24,12 +24,15 @@ import (
"log"
"os"
"path/filepath"
+ "strconv"
"strings"
"github.com/apache/beam/sdks/go/pkg/beam/artifact"
+ pb "github.com/apache/beam/sdks/go/pkg/beam/model/fnexecution_v1"
"github.com/apache/beam/sdks/go/pkg/beam/provision"
"github.com/apache/beam/sdks/go/pkg/beam/util/execx"
"github.com/apache/beam/sdks/go/pkg/beam/util/grpcx"
+ "github.com/apache/beam/sdks/go/pkg/beam/util/syscallx"
)
var (
@@ -76,7 +79,8 @@ func main() {
log.Fatalf("Failed to convert pipeline options: %v", err)
}
- // (2) Retrieve the staged user jars.
+ // (2) Retrieve the staged user jars. We ignore any disk limit,
+ // because the staged jars are mandatory.
dir := filepath.Join(*semiPersistDir, "staged")
@@ -102,10 +106,27 @@ func main() {
}
args := []string{
+ "-Xmx" + strconv.FormatUint(heapSizeLimit(info), 10),
"-cp", strings.Join(cp, ":"),
"org.apache.beam.fn.harness.FnHarness",
}
+
log.Printf("Executing: java %v", strings.Join(args, " "))
log.Fatalf("Java exited: %v", execx.Execute("java", args...))
}
+
+// heapSizeLimit returns 80% of the runner limit, if provided. If not provided,
+// it returns 70% of the physical memory on the machine. If it cannot determine
+// that value, it returns 1GB. This is an imperfect heuristic. I aims to
+// ensure there is memory for non-heap use and other overhead, while also not
+// underutilizing the machine.
+func heapSizeLimit(info *pb.ProvisionInfo) uint64 {
+ if provided := info.GetResourceLimits().GetMemory().GetSize(); provided > 0 {
+ return (provided * 80) / 100
+ }
+ if size, err := syscallx.PhysicalMemorySize(); err == nil {
+ return (size * 70) / 100
+ }
+ return 1 << 30
+}