# Copyright 2012 the V8 project authors. All rights reserved.
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are
# met:
#
# * Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
# * Redistributions in binary form must reproduce the above
# copyright notice, this list of conditions and the following
# disclaimer in the documentation and/or other materials provided
# with the distribution.
# * Neither the name of Google Inc. nor the names of its
# contributors may be used to endorse or promote products derived
# from this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
import os
import socket
import subprocess
import threading
import time
from . import distro
from ..local import execution
from ..local import perfdata
from ..objects import peer
from ..objects import workpacket
from ..server import compression
from ..server import constants
from ..server import local_handler
from ..server import signatures
def GetPeers():
data = local_handler.LocalQuery([constants.REQUEST_PEERS])
if not data: return []
return [ peer.Peer.Unpack(p) for p in data ]
class NetworkedRunner(execution.Runner):
def __init__(self, suites, progress_indicator, context, peers, workspace):
self.suites = suites
datapath = os.path.join("out", "testrunner_data")
# TODO(machenbach): These fields should exist now in the superclass.
# But there is no super constructor call. Check if this is a problem.
self.perf_data_manager = perfdata.PerfDataManager(datapath)
self.perfdata = self.perf_data_manager.GetStore(context.arch, context.mode)
for s in suites:
for t in s.tests:
t.duration = self.perfdata.FetchPerfData(t) or 1.0
self._CommonInit(suites, progress_indicator, context)
self.tests = [] # Only used if we need to fall back to local execution.
self.tests_lock = threading.Lock()
self.peers = peers
self.pubkey_fingerprint = None # Fetched later.
self.base_rev = subprocess.check_output(
"cd %s; git log -1 --format=%%H --grep=git-svn-id" % workspace,
shell=True).strip()
self.base_svn_rev = subprocess.check_output(
"cd %s; git log -1 %s" # Get commit description.
" | grep -e '^\s*git-svn-id:'" # Extract "git-svn-id" line.
" | awk '{print $2}'" # Extract "repository@revision" part.
" | sed -e 's/.*@//'" % # Strip away "repository@".
(workspace, self.base_rev), shell=True).strip()
self.patch = subprocess.check_output(
"cd %s; git diff %s" % (workspace, self.base_rev), shell=True)
self.binaries = {}
self.initialization_lock = threading.Lock()
self.initialization_lock.acquire() # Released when init is done.
self._OpenLocalConnection()
self.local_receiver_thread = threading.Thread(
target=self._ListenLocalConnection)
self.local_receiver_thread.daemon = True
self.local_receiver_thread.start()
self.initialization_lock.acquire()
self.initialization_lock.release()
def _OpenLocalConnection(self):
self.local_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
code = self.local_socket.connect_ex(("localhost", constants.CLIENT_PORT))
if code != 0:
raise RuntimeError("Failed to connect to local server")
compression.Send([constants.REQUEST_PUBKEY_FINGERPRINT], self.local_socket)
def _ListenLocalConnection(self):
release_lock_countdown = 1 # Pubkey.
self.local_receiver = compression.Receiver(self.local_socket)
while not self.local_receiver.IsDone():
data = self.local_receiver.Current()
if data[0] == constants.REQUEST_PUBKEY_FINGERPRINT:
pubkey = data[1]
if not pubkey: raise RuntimeError("Received empty public key")
self.pubkey_fingerprint = pubkey
release_lock_countdown -= 1
if release_lock_countdown == 0:
self.initialization_lock.release()
release_lock_countdown -= 1 # Prevent repeated triggering.
self.local_receiver.Advance()
def Run(self, jobs):
self.indicator.Starting()
need_libv8 = False
for s in self.suites:
shell = s.shell()
if shell not in self.binaries:
path = os.path.join(self.context.shell_dir, shell)
# Check if this is a shared library build.
try:
ldd = subprocess.check_output("ldd %s | grep libv8\\.so" % (path),
shell=True)
ldd = ldd.strip().split(" ")
assert ldd[0] == "libv8.so"
assert ldd[1] == "=>"
need_libv8 = True
binary_needs_libv8 = True
libv8 = signatures.ReadFileAndSignature(ldd[2])
except:
binary_needs_libv8 = False
binary = signatures.ReadFileAndSignature(path)
if binary[0] is None:
print("Error: Failed to create signature.")
assert binary[1] != 0
return binary[1]
binary.append(binary_needs_libv8)
self.binaries[shell] = binary
if need_libv8:
self.binaries["libv8.so"] = libv8
distro.Assign(self.suites, self.peers)
# Spawn one thread for each peer.
threads = []
for p in self.peers:
thread = threading.Thread(target=self._TalkToPeer, args=[p])
threads.append(thread)
thread.start()
try:
for thread in threads:
# Use a timeout so that signals (Ctrl+C) will be processed.
thread.join(timeout=10000000)
self._AnalyzePeerRuntimes()
except KeyboardInterrupt:
self.terminate = True
raise
except Exception, _e:
# If there's an exception we schedule an interruption for any
# remaining threads...
self.terminate = True
# ...and then reraise the exception to bail out.
raise
compression.Send(constants.END_OF_STREAM, self.local_socket)
self.local_socket.close()
if self.tests:
self._RunInternal(jobs)
self.indicator.Done()
return not self.failed
def _TalkToPeer(self, peer):
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(self.context.timeout + 10)
code = sock.connect_ex((peer.address, constants.PEER_PORT))
if code == 0:
try:
peer.runtime = None
start_time = time.time()
packet = workpacket.WorkPacket(peer=peer, context=self.context,
base_revision=self.base_svn_rev,
patch=self.patch,
pubkey=self.pubkey_fingerprint)
data, test_map = packet.Pack(self.binaries)
compression.Send(data, sock)
compression.Send(constants.END_OF_STREAM, sock)
rec = compression.Receiver(sock)
while not rec.IsDone() and not self.terminate:
data_list = rec.Current()
for data in data_list:
test_id = data[0]
if test_id < 0:
# The peer is reporting an error.
with self.lock:
print("\nPeer %s reports error: %s" % (peer.address, data[1]))
continue
test = test_map.pop(test_id)
test.MergeResult(data)
try:
self.perfdata.UpdatePerfData(test)
except Exception, e:
print("UpdatePerfData exception: %s" % e)
pass # Just keep working.
with self.lock:
perf_key = self.perfdata.GetKey(test)
compression.Send(
[constants.INFORM_DURATION, perf_key, test.duration,
self.context.arch, self.context.mode],
self.local_socket)
has_unexpected_output = test.suite.HasUnexpectedOutput(test)
if has_unexpected_output:
self.failed.append(test)
if test.output.HasCrashed():
self.crashed += 1
else:
self.succeeded += 1
self.remaining -= 1
self.indicator.HasRun(test, has_unexpected_output)
rec.Advance()
peer.runtime = time.time() - start_time
except KeyboardInterrupt:
sock.close()
raise
except Exception, e:
print("Got exception: %s" % e)
pass # Fall back to local execution.
else:
compression.Send([constants.UNRESPONSIVE_PEER, peer.address],
self.local_socket)
sock.close()
if len(test_map) > 0:
# Some tests have not received any results. Run them locally.
print("\nNo results for %d tests, running them locally." % len(test_map))
self._EnqueueLocally(test_map)
def _EnqueueLocally(self, test_map):
with self.tests_lock:
for test in test_map:
self.tests.append(test_map[test])
def _AnalyzePeerRuntimes(self):
total_runtime = 0.0
total_work = 0.0
for p in self.peers:
if p.runtime is None:
return
total_runtime += p.runtime
total_work += p.assigned_work
for p in self.peers:
p.assigned_work /= total_work
p.runtime /= total_runtime
perf_correction = p.assigned_work / p.runtime
old_perf = p.relative_performance
p.relative_performance = (old_perf + perf_correction) / 2.0
compression.Send([constants.UPDATE_PERF, p.address,
p.relative_performance],
self.local_socket)