mirror of
https://gitlab.com/ita1024/waf.git
synced 2024-11-22 09:57:15 +01:00
Additional notes on preforked build processes
This commit is contained in:
Thomas Nagy
parent
8d9e7fda30
commit
fa8d113d41
221
playground/prefork/Prefork.java
Normal file
221
playground/prefork/Prefork.java
Normal file
@ -0,0 +1,221 @@
|
||||
// Thomas Nagy, 2015
|
||||
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
import java.util.Arrays;
|
||||
import java.util.List;
|
||||
import java.util.Date;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Comparator;
|
||||
import java.util.Collections;
|
||||
import java.util.Scanner;
|
||||
import java.lang.Math;
|
||||
import java.lang.StringBuilder;
|
||||
import java.io.IOException;
|
||||
import java.io.InputStream;
|
||||
import java.io.OutputStream;
|
||||
import java.io.File;
|
||||
import java.io.StringWriter;
|
||||
import java.io.PrintWriter;
|
||||
|
||||
import java.lang.Math;
|
||||
import com.eclipsesource.json.JsonObject;
|
||||
import com.eclipsesource.json.JsonArray;
|
||||
|
||||
import java.lang.ProcessBuilder;
|
||||
import java.lang.ProcessBuilder.Redirect;
|
||||
import java.lang.Process;
|
||||
|
||||
import java.io.FileOutputStream;
|
||||
import java.net.ServerSocket;
|
||||
import java.net.Socket;
|
||||
|
||||
public class Prefork implements Runnable, Comparator<Object[]> {
|
||||
private static int HEADER_SIZE = 64;
|
||||
private static int BUF = 2048;
|
||||
static String SHARED_KEY = "";
|
||||
|
||||
private Socket sock = null;
|
||||
private int port = 0;
|
||||
|
||||
public Prefork(Socket sock, int port) {
|
||||
this.sock = sock;
|
||||
this.port = port;
|
||||
}
|
||||
|
||||
public boolean safeCompare(String a, String b) {
|
||||
int sum = Math.abs(a.length() - b.length());
|
||||
for (int i = 0; i < b.length(); ++i) {
|
||||
sum |= a.charAt(i) ^ b.charAt(i);
|
||||
}
|
||||
return sum == 0;
|
||||
}
|
||||
|
||||
public void run ()
|
||||
{
|
||||
try {
|
||||
if (sock != null)
|
||||
{
|
||||
while (true) {
|
||||
InputStream in = sock.getInputStream();
|
||||
OutputStream out = sock.getOutputStream();
|
||||
|
||||
byte b[] = new byte[HEADER_SIZE];
|
||||
int off = 0;
|
||||
while (off < b.length) {
|
||||
off += in.read(b, off, b.length - off);
|
||||
}
|
||||
|
||||
String line = new String(b);
|
||||
String key = line.substring(line.length() - 20);
|
||||
if (key.length() != 20) {
|
||||
System.err.println("Fatal error in the application");
|
||||
}
|
||||
if (!safeCompare(key, SHARED_KEY))
|
||||
{
|
||||
System.err.println("Invalid key given " + key.length() + " " + SHARED_KEY.length() + " " + key + " " + SHARED_KEY);
|
||||
sock.close();
|
||||
}
|
||||
|
||||
//System.out.println(new String(b));
|
||||
String[] args = line.substring(0, line.length() - 20).split(",");
|
||||
if (args[0].equals("REQ")) {
|
||||
process(args, sock);
|
||||
}
|
||||
else
|
||||
{
|
||||
System.out.println("Invalid command " + new String(b) + " on port " + this.port);
|
||||
sock.close();
|
||||
break;
|
||||
}
|
||||
}
|
||||
} else {
|
||||
// magic trick to avoid creating a new inner class
|
||||
ServerSocket server = new ServerSocket(port);
|
||||
server.setReuseAddress(true);
|
||||
while (true) {
|
||||
Prefork tmp = new Prefork(server.accept(), port);
|
||||
Thread t = new Thread(tmp);
|
||||
t.start();
|
||||
}
|
||||
}
|
||||
} catch (InterruptedException e) {
|
||||
e.printStackTrace();
|
||||
} catch (IOException e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
}
|
||||
|
||||
public String make_out(Socket sock, String stdout, String stderr, String exc) {
|
||||
if ((stdout == null || stdout.length() == 0) && (stderr == null || stderr.length() == 0) && (exc == null || exc.length() == 0))
|
||||
{
|
||||
return "";
|
||||
}
|
||||
|
||||
JsonArray ret = new JsonArray();
|
||||
ret.add(stdout);
|
||||
ret.add(stderr);
|
||||
ret.add(exc);
|
||||
return ret.toString();
|
||||
}
|
||||
|
||||
public String readFile(File f) throws IOException {
|
||||
String ret = new Scanner(f).useDelimiter("\\A").next();
|
||||
return ret;
|
||||
}
|
||||
|
||||
public void process(String[] args, Socket sock) throws IOException, InterruptedException {
|
||||
long size = new Long(args[1].trim());
|
||||
//System.out.println("" + args[1] + " " + args[2] + " " + args[3] + " " + args.length);
|
||||
|
||||
byte[] buf = new byte[BUF];
|
||||
StringBuilder sb = new StringBuilder();
|
||||
InputStream in = sock.getInputStream();
|
||||
long cnt = 0;
|
||||
while (cnt < size) {
|
||||
int c = in.read(buf, 0, (int) Math.min(BUF, size-cnt));
|
||||
if (c == 0) {
|
||||
throw new RuntimeException("Connection closed too early");
|
||||
}
|
||||
sb.append(new String(buf, 0, c));
|
||||
cnt += c;
|
||||
}
|
||||
|
||||
String stdout = null;
|
||||
String stderr = null;
|
||||
String exc = null;
|
||||
|
||||
JsonObject kw = JsonObject.readFrom(sb.toString());
|
||||
boolean isShell = kw.get("shell").asBoolean();
|
||||
|
||||
String[] command = null;
|
||||
|
||||
if (isShell) {
|
||||
command = new String[] {"sh", "-c", kw.get("cmd").asString()};
|
||||
}
|
||||
else
|
||||
{
|
||||
JsonArray arr = kw.get("cmd").asArray();
|
||||
int siz = arr.size();
|
||||
command = new String[siz];
|
||||
for (int i =0; i < siz; ++i) {
|
||||
command[i] = arr.get(i).asString();
|
||||
}
|
||||
}
|
||||
|
||||
ProcessBuilder pb = new ProcessBuilder(command);
|
||||
String cwd = kw.get("cwd").asString();
|
||||
if (cwd != null) {
|
||||
pb.directory(new File(cwd));
|
||||
}
|
||||
|
||||
long threadId = Thread.currentThread().getId();
|
||||
String errFile = "log_err_" + threadId;
|
||||
File elog = new File(errFile);
|
||||
pb.redirectError(Redirect.to(elog));
|
||||
|
||||
String outFile = "log_out_" + threadId;
|
||||
File olog = new File(outFile);
|
||||
pb.redirectOutput(Redirect.to(olog));
|
||||
|
||||
int status = -8;
|
||||
try {
|
||||
Process p = pb.start();
|
||||
status = p.waitFor();
|
||||
} catch (IOException e) {
|
||||
StringWriter sw = new StringWriter();
|
||||
e.printStackTrace(new PrintWriter(sw));
|
||||
exc = sw.toString();
|
||||
}
|
||||
|
||||
if (olog.length() != 0) {
|
||||
stdout = readFile(olog);
|
||||
}
|
||||
|
||||
if (elog.length() != 0) {
|
||||
stderr = readFile(elog);
|
||||
}
|
||||
OutputStream out = sock.getOutputStream();
|
||||
String msg = make_out(sock, stdout, stderr, exc);
|
||||
|
||||
// RES, status, ret size
|
||||
String ret = String.format("%-64s", String.format("RES,%d,%d", status, msg.length()));
|
||||
out.write(ret.getBytes());
|
||||
if (msg.length() > 0)
|
||||
{
|
||||
out.write(msg.getBytes());
|
||||
}
|
||||
}
|
||||
|
||||
public int compare(Object[] a, Object[] b) {
|
||||
return ((Long) a[0]).compareTo((Long) b[0]);
|
||||
}
|
||||
|
||||
public static void main(String[] args) {
|
||||
Map<String, String> env = System.getenv();
|
||||
SHARED_KEY = env.get("SHARED_KEY");
|
||||
Prefork tmp = new Prefork(null, new Integer(args[0]));
|
||||
tmp.run();
|
||||
}
|
||||
}
|
||||
|
||||
7
playground/prefork/README
Normal file
7
playground/prefork/README
Normal file
@ -0,0 +1,7 @@
|
||||
TODO - work in progress
|
||||
|
||||
Get a copy of https://github.com/ralfstx/minimal-json, currently minimal-json-0.9.3-SNAPSHOT.jar
|
||||
|
||||
Build the java file with:
|
||||
javac -target 1.7 -source 1.7 -cp ./minimal-json-0.9.3-SNAPSHOT.jar:. Prefork.java
|
||||
|
||||
205
playground/prefork/preforkjava.py
Executable file
205
playground/prefork/preforkjava.py
Executable file
@ -0,0 +1,205 @@
|
||||
#! /usr/bin/env python
|
||||
# encoding: utf-8
|
||||
# Thomas Nagy, 2015 (ita)
|
||||
|
||||
import os, re, socket, threading, sys, subprocess, time, atexit, traceback, random
|
||||
try:
|
||||
import SocketServer
|
||||
except ImportError:
|
||||
import socketserver as SocketServer
|
||||
try:
|
||||
from queue import Queue
|
||||
except ImportError:
|
||||
from Queue import Queue
|
||||
|
||||
import json as pickle
|
||||
|
||||
DEFAULT_PORT = 51200
|
||||
SHARED_KEY = None
|
||||
HEADER_SIZE = 64
|
||||
|
||||
REQ = 'REQ'
|
||||
RES = 'RES'
|
||||
BYE = 'BYE'
|
||||
|
||||
def make_header(params, cookie=''):
|
||||
header = ','.join(params)
|
||||
header = header.ljust(HEADER_SIZE - len(cookie))
|
||||
assert(len(header) == HEADER_SIZE - len(cookie))
|
||||
header = header + cookie
|
||||
if sys.hexversion > 0x3000000:
|
||||
header = header.encode('iso8859-1')
|
||||
return header
|
||||
|
||||
if 1:
|
||||
from waflib import Logs, Utils, Runner, Errors
|
||||
|
||||
def init_task_pool(self):
|
||||
# lazy creation, and set a common pool for all task consumers
|
||||
pool = self.pool = []
|
||||
for i in range(self.numjobs):
|
||||
consumer = Runner.get_pool()
|
||||
pool.append(consumer)
|
||||
consumer.idx = i
|
||||
self.ready = Queue(0)
|
||||
def setq(consumer):
|
||||
consumer.ready = self.ready
|
||||
try:
|
||||
threading.current_thread().idx = consumer.idx
|
||||
except Exception as e:
|
||||
print(e)
|
||||
for x in pool:
|
||||
x.ready.put(setq)
|
||||
return pool
|
||||
Runner.Parallel.init_task_pool = init_task_pool
|
||||
|
||||
PORT = 51200
|
||||
|
||||
def make_server(bld, idx):
|
||||
wd = os.path.dirname(os.path.abspath('__file__'))
|
||||
port = PORT + idx
|
||||
cmd = "java -cp %s/minimal-json-0.9.3-SNAPSHOT.jar:. Prefork %d" % (wd, PORT)
|
||||
proc = subprocess.Popen(cmd.split(), shell=False, cwd=d)
|
||||
proc.port = port
|
||||
return proc
|
||||
|
||||
def make_conn(bld, srv):
|
||||
#port = PORT + idx
|
||||
port = srv.port
|
||||
conn = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
conn.connect(('127.0.0.1', port))
|
||||
return conn
|
||||
|
||||
SERVERS = []
|
||||
CONNS = []
|
||||
def close_all():
|
||||
global SERVERS
|
||||
while SERVERS:
|
||||
srv = SERVERS.pop()
|
||||
pid = srv.pid
|
||||
try:
|
||||
srv.kill()
|
||||
except Exception as e:
|
||||
pass
|
||||
atexit.register(close_all)
|
||||
|
||||
def put_data(conn, data):
|
||||
conn.send(data)
|
||||
|
||||
def read_data(conn, siz):
|
||||
ret = conn.recv(siz)
|
||||
#if not ret:
|
||||
# print("closed connection?")
|
||||
assert(len(ret) == siz)
|
||||
return ret
|
||||
|
||||
def exec_command(self, cmd, **kw):
|
||||
|
||||
if 'stdout' in kw:
|
||||
if kw['stdout'] not in (None, subprocess.PIPE):
|
||||
return self.exec_command_old(cmd, **kw)
|
||||
elif 'stderr' in kw:
|
||||
if kw['stderr'] not in (None, subprocess.PIPE):
|
||||
return self.exec_command_old(cmd, **kw)
|
||||
|
||||
kw['shell'] = isinstance(cmd, str)
|
||||
Logs.debug('runner: %r' % cmd)
|
||||
Logs.debug('runner_env: kw=%s' % kw)
|
||||
|
||||
if self.logger:
|
||||
self.logger.info(cmd)
|
||||
|
||||
if 'stdout' not in kw:
|
||||
kw['stdout'] = subprocess.PIPE
|
||||
if 'stderr' not in kw:
|
||||
kw['stderr'] = subprocess.PIPE
|
||||
|
||||
if Logs.verbose and not kw['shell'] and not Utils.check_exe(cmd[0]):
|
||||
raise Errors.WafError("Program %s not found!" % cmd[0])
|
||||
|
||||
idx = threading.current_thread().idx
|
||||
kw['cmd'] = cmd
|
||||
|
||||
data = pickle.dumps(kw)
|
||||
params = [REQ, str(len(data))]
|
||||
header = make_header(params, self.SHARED_KEY)
|
||||
|
||||
conn = CONNS[idx]
|
||||
|
||||
put_data(conn, header)
|
||||
put_data(conn, data)
|
||||
|
||||
data = read_data(conn, HEADER_SIZE)
|
||||
if sys.hexversion > 0x3000000:
|
||||
data = data.decode('iso8859-1')
|
||||
|
||||
#print("received %r" % data)
|
||||
lst = data.split(',')
|
||||
ret = int(lst[1])
|
||||
dlen = int(lst[2])
|
||||
|
||||
out = err = None
|
||||
if dlen:
|
||||
data = read_data(conn, dlen)
|
||||
(out, err, exc) = pickle.loads(data)
|
||||
if exc:
|
||||
raise Errors.WafError('Execution failure: %s' % exc)
|
||||
|
||||
if out:
|
||||
if not isinstance(out, str):
|
||||
out = out.decode(sys.stdout.encoding or 'iso8859-1')
|
||||
if self.logger:
|
||||
self.logger.debug('out: %s' % out)
|
||||
else:
|
||||
Logs.info(out, extra={'stream':sys.stdout, 'c1': ''})
|
||||
if err:
|
||||
if not isinstance(err, str):
|
||||
err = err.decode(sys.stdout.encoding or 'iso8859-1')
|
||||
if self.logger:
|
||||
self.logger.error('err: %s' % err)
|
||||
else:
|
||||
Logs.info(err, extra={'stream':sys.stderr, 'c1': ''})
|
||||
|
||||
return ret
|
||||
|
||||
def init_key(ctx):
|
||||
try:
|
||||
key = ctx.SHARED_KEY = os.environ['SHARED_KEY']
|
||||
except KeyError:
|
||||
key = "".join([chr(random.SystemRandom().randint(40, 126)) for x in range(20)])
|
||||
os.environ['SHARED_KEY'] = ctx.SHARED_KEY = key
|
||||
return key
|
||||
|
||||
def init_servers(ctx, maxval):
|
||||
while len(SERVERS) < 1:
|
||||
i = len(SERVERS)
|
||||
srv = make_server(ctx, i)
|
||||
SERVERS.append(srv)
|
||||
while len(CONNS) < maxval:
|
||||
i = len(CONNS)
|
||||
srv = SERVERS[0]
|
||||
conn = None
|
||||
for x in range(30):
|
||||
try:
|
||||
conn = make_conn(ctx, srv)
|
||||
break
|
||||
except socket.error:
|
||||
time.sleep(0.01)
|
||||
if not conn:
|
||||
raise ValueError('Could not start the server!')
|
||||
CONNS.append(conn)
|
||||
|
||||
def options(opt):
|
||||
init_key(opt)
|
||||
init_servers(opt, 40)
|
||||
|
||||
def build(bld):
|
||||
if bld.cmd == 'clean':
|
||||
return
|
||||
|
||||
init_key(bld)
|
||||
init_servers(bld, bld.jobs)
|
||||
|
||||
bld.__class__.exec_command_old = bld.__class__.exec_command
|
||||
bld.__class__.exec_command = exec_command
|
||||
|
||||
@ -1,4 +1,4 @@
|
||||
#!/usr/bin/python
|
||||
#!/usr/bin/env python
|
||||
# encoding: utf-8
|
||||
|
||||
import sys
|
||||
|
||||
@ -57,8 +57,10 @@ def make_header(params, cookie=''):
|
||||
return header
|
||||
|
||||
def safe_compare(x, y):
|
||||
vec = [abs(ord(a) - ord(b)) for (a, b) in zip(x, y)]
|
||||
return not sum(vec)
|
||||
sum = 0
|
||||
for (a, b) in zip(x, y):
|
||||
sum |= ord(a) ^ ord(b)
|
||||
return sum == 0
|
||||
|
||||
re_valid_query = re.compile('^[a-zA-Z0-9_, ]+$')
|
||||
class req(SocketServer.StreamRequestHandler):
|
||||
|
||||
Loading…
Reference in New Issue
Block a user