af1f78b80e
This removes the "AboutToRun" callback. It makes not much sense as it is not actually called before a test is run, but just before the results are retrieved from the results queue. Therefore, from an information-value p-o-v it is redundant to the "HasRun" callback. This also reduces log size on the bots by 50%, in order to reduce buildbot slowness, which is tied to the amount of logs. Review URL: https://codereview.chromium.org/1874973003 Cr-Commit-Position: refs/heads/master@{#35382}
254 lines
10 KiB
Python
254 lines
10 KiB
Python
# 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)
|