pydisco/disco/comm.py test/test_partfile.py @@ -9,4 +9,4 @@ DISCO_USER=$USER DISCO_LOG=$DISCO DISCO_PID_DIR=$DISCO DISCO_MASTER_PORT=7000 - +#DISCO_FLAGS=resultfs conf/disco.conf.dev-example @@ -1,12 +1,12 @@ -module(disco_config). --export([get_config_table/0, save_config_table/1]). +-export([get_config_table/0, save_config_table/1, expand_range/2]). config_file() -> {ok, Val} = application:get_env(disco_config), Val. -add_nodes([FirstNode, Max], Instances) -> +expand_range(FirstNode, Max) -> Len = string:len(FirstNode), FieldLen = string:len(Max), MaxNum = list_to_integer(Max), @@ -14,9 +14,11 @@ add_nodes([FirstNode, Max], Instances) -> MinNum = list_to_integer( string:sub_string(FirstNode, Len - FieldLen + 1)), Format = lists:flatten(io_lib:format("~s~~.~w.0w", [Name, FieldLen])), + [lists:flatten(io_lib:fwrite(Format, [I])) || + I <- lists:seq(MinNum, MaxNum)]. - [{lists:flatten(io_lib:fwrite(Format, [I])), Instances} || - I <- lists:seq(MinNum, MaxNum)]; +add_nodes([FirstNode, Max], Instances) -> + [{N, Instances} || N <- expand_range(FirstNode, Max)]; add_nodes([Node], Instances) -> {Node, Instances}. master/src/disco_config.erl @@ -177,8 +177,8 @@ handle_call({purge_job, JobName}, From, State) -> {ok, Root} = application:get_env(disco_root), handle_call({clean_job, JobName}, From, State), Nodes = [lists:flatten( - ["dir://", Node, "/map/", - Node, "/", jobhome(JobName)]) || + ["dir://", Node, "/", Node, "/", + jobhome(JobName), "/null"]) || {Node, _} <- ets:tab2list(node_load)], garbage_collect:remove_job(Nodes), garbage_collect:remove_dir(filename:join([Root, jobhome(JobName)])) master/src/disco_server.erl @@ -1,16 +1,18 @@ -module(garbage_collect). --export([remove_map_results/1, remove_job/1, remove_dir/1]). +-export([remove_map_results/1, remove_job/1, remove_dir/1, move_results/1]). + +-define(RESULTFS_TIMEOUT, 300000). % 5min spawn_remote([], _) -> ok; spawn_remote([Url|Urls], F) when is_binary(Url) -> spawn_remote([binary_to_list(Url)|Urls], F); spawn_remote([Url|Urls], F) when is_list(Url) -> - ["dir:", Node, _|S0] = string:tokens(Url, "/"), - S = filename:join(S0), + ["dir:", Node, _, Pref, JobName, File] = string:tokens(Url, "/"), + JobRoot = filename:join([Node, Pref, JobName]), SName = disco_worker:slave_name(Node), case net_adm:ping(SName) of - pong -> spawn(SName, fun () -> F(S, Url) end); + pong -> spawn(SName, fun () -> F(File, JobRoot, Node) end); _ -> ok end, spawn_remote(Urls, F). @@ -21,18 +23,62 @@ spawn_remote([Url|Urls], F) when is_list(Url) -> remove_dir(Dir) -> spawn(fun() -> os:cmd("rm -Rf " ++ Dir) end). -remove_map_results(Urls) -> - spawn_remote(Urls, fun (JobName, Url) -> +remove_job(Urls) -> + spawn_remote(Urls, fun (_, JobRoot, Node) -> Root = os:getenv("DISCO_ROOT"), - error_logger:info_report({"Deleting map results at", Url}), - os:cmd("rm -Rf " ++ - filename:join([Root, "data", JobName, "map-*"])) + error_logger:info_report({"Deleting all job files at", Node, JobRoot}), + remove_dir(filename:join([Root, "data", JobRoot])), + remove_dir(filename:join([Root, "temp", JobRoot])) end). -remove_job(Urls) -> - spawn_remote(Urls, fun (JobName, Url) -> +remove_map_results(Urls) -> + spawn_remote(Urls, fun (File, JobRoot, _) -> Root = os:getenv("DISCO_ROOT"), - error_logger:info_report({"Deleting all job files at", Url}), - remove_dir(filename:join([Root, "data", JobName])) + lists:foreach(fun(F) -> + C = "rm -Rf " ++ + filename:join([Root, "data", JobRoot, F]), + os:cmd(C) + end, parse_dir(File, JobRoot)) end). - + +move_results(Urls) -> + Parent = self(), + spawn_remote(Urls, fun (File, JobRoot, Node) -> + Root = os:getenv("DISCO_ROOT"), + Src = filename:join([Root, "temp", JobRoot]), + Dst = filename:join([Root, "data", JobRoot]), + os:cmd("mkdir -p " ++ Dst ++ "/oob"), + lists:foreach(fun(F) -> + SrcF = filename:join(Src, F), + X = os:cmd("mv " ++ SrcF ++ " " ++ Dst), + if X =/= [] -> + Parent ! {resultfs_error, Node, X}; + true -> ok + end + end, parse_dir(File, JobRoot)), + os:cmd("mv " ++ Src ++ "/oob/* " ++ Dst ++ "/oob"), + Parent ! {resultfs_node_ok, Node} + end), + wait_nodes(length(Urls)). + +wait_nodes(0) -> ok; +wait_nodes(N) -> + receive + {resultfs_node_ok, _} -> + wait_nodes(N - 1); + {resultfs_error, Node, Error} -> + {error, Node, Error}; + _ -> + wait_nodes(N) + after ?RESULTFS_TIMEOUT -> + timeout + end. + +parse_dir(File, JobRoot) -> + Root = os:getenv("DISCO_ROOT"), + case string:tokens(File, ":") of + [B, N] -> disco_config:expand_range(B, N); + [Index] -> + IFile = filename:join([Root, "temp", JobRoot, File]), + [Index|string:tokens(os:cmd("cat " ++ IFile), "\n")] + end. master/src/garbage_collect.erl @@ -225,6 +225,37 @@ check_failure_rate(Name, PartID, Mode, L) -> ok end. +kill_job(Name, Msg, P, Type) -> + event_server:event(Name, Msg, P, []), + gen_server:call(disco_server, {kill_job, Name}), + gen_server:cast(event_server, {flush_events, Name}), + exit(Type). + +resultfs_enabled() -> + S = case os:getenv("DISCO_FLAGS") of + false -> []; + X -> string:tokens(string:to_lower(X), " ") + end, + [ 1 || "resultfs" <- S] =/= []. + +move_to_resultfs(_, _, false) -> ok; +move_to_resultfs(Name, R, _) -> + event_server:event(Name, "Moving results to resultfs", [], []), + case catch garbage_collect:move_results(R) of + ok -> ok; + {error, Node, Error} -> + kill_job(Name, + "ERROR: Moving to resultfs failed on ~s: ~s", + [Node, Error], logged_error); + timeout -> + kill_job(Name, + "ERROR: Moving to resultfs failed (timeout)", + [], logged_error); + Error -> + kill_job(Name, + "ERROR: Moving to resultfs failed: ~p", + [Error], unknown_error) + end. % run_task() is a common supervisor for both the map and reduce tasks. % Its main function is to catch and report any errors that occur during @@ -239,21 +270,17 @@ run_task(Inputs, Mode, Name, MaxN) -> case catch work(Inputs, Mode, Name, 0, MaxN, {Results, Failures}) of ok -> ok; logged_error -> - event_server:event(Name, + kill_job(Name, "ERROR: Job terminated due to the previous errors", - [], []), - gen_server:call(disco_server, {kill_job, Name}), - gen_server:cast(event_server, {flush_events, Name}), - exit(logged_error); + [], logged_error); Error -> - event_server:event(Name, + kill_job(Name, "ERROR: Job coordinator failed unexpectedly: ~p", - [Error], []), - gen_server:call(disco_server, {kill_job, Name}), - gen_server:cast(event_server, {flush_events, Name}), - exit(unknown_error) + [Error], unknown_error) end, + R = [list_to_binary(X) || {X, _} <- ets:tab2list(Results)], + move_to_resultfs(Name, R, resultfs_enabled()), ets:delete(Results), ets:delete(Failures), R. master/src/handle_job.erl @@ -2,16 +2,11 @@ import os, sys from disconode import disco_worker as dw -from disconode import external -from disconode.util import ensure_path - -from disconode.disco_worker import\ - HTTP_PORT, LOCAL_PATH, PARAMS_FILE, EXT_MAP, EXT_REDUCE,\ - MAP_OUTPUT, CHUNK_OUTPUT, REDUCE_DL, REDUCE_SORTED, REDUCE_OUTPUT,\ - OOB_FILE +from disconode.util import ensure_path, ensure_file +from disconode.disco_worker import JOB_ROOT, OOB_FILE, PARAMS_FILE from disco.netstring import decode_netstring_fd -from disco import util +from disco import util, comm if __name__ == "__main__": if len(sys.argv) < 7: @@ -28,12 +23,19 @@ if __name__ == "__main__": master_url = sys.argv[4] ensure_path(OOB_FILE % "", False) - + try: - url = "%s/params" % master_url - external.ensure_file(PARAMS_FILE, url = url, mode = 444) + if util.resultfs_enabled: + x, x, root = util.load_conf() + url = "%s/data/%s/params" % (root, "/".join(\ + master_url.strip("/").split("/")[-3:])) + fn = lambda: file(url).read() + else: + url = "%s/params" % master_url + fn = lambda: comm.download(url) + ensure_file(PARAMS_FILE, fn, mode = 444) except Exception, e: - util.data_err("Failed to get %s" % url, master_url) + util.data_err("Failed to get %s: %s" % (url, e), master_url) try: m = decode_netstring_fd(file(PARAMS_FILE)) node/disco-worker @@ -1,15 +1,12 @@ import os, subprocess, cStringIO, marshal, time, sys, cPickle, md5 import re, traceback, tempfile, struct, random -from disco.util import parse_dir, load_conf, err, data_err, msg +from disco.util import\ + parse_dir, load_conf, err, data_err, msg, resultfs_enabled, load_oob from disco.func import re_reader, netstr_reader from disco.netstring import * from disconode.util import * from disconode import external - -try: - import disco.comm_curl as comm -except: - import disco.comm_httplib as comm +from disco import comm oob_chars = re.compile("[^a-zA-Z_\-:0-9]") @@ -60,30 +57,36 @@ def this_inputs(): return sys.argv[6:] def init(): - global HTTP_PORT, LOCAL_PATH, PARAMS_FILE, EXT_MAP, EXT_REDUCE,\ - PART_SUFFIX, MAP_OUTPUT, CHUNK_OUTPUT, REDUCE_DL,\ - REDUCE_SORTED, REDUCE_OUTPUT, OOB_FILE, OOB_URL,\ - JOB_HOME + global HTTP_PORT, PARAMS_FILE, EXT_MAP, EXT_REDUCE,\ + PART_SUFFIX, MAP_OUTPUT, REDUCE_DL,\ + REDUCE_SORTED, REDUCE_OUTPUT, OOB_FILE,\ + JOB_HOME, DISCO_ROOT, JOB_ROOT, PART_OUTPUT,\ + MAP_INDEX, REDUCE_INDEX - tmp, HTTP_PORT, LOCAL_PATH = load_conf() + tmp, HTTP_PORT, DISCO_ROOT = load_conf() job_name = this_name() JOB_HOME = "%s/%s/%s/" %\ (this_host(), md5.md5(job_name).hexdigest()[:2], job_name) - pp = LOCAL_PATH + "/" + JOB_HOME + JOB_ROOT = "%s/data/%s" % (DISCO_ROOT, JOB_HOME) + + if resultfs_enabled: + pp = "%s/temp/%s" % (DISCO_ROOT, JOB_HOME) + else: + pp = JOB_ROOT - OOB_URL = ("http://%s/disco/ctrl/oob_get?" % this_master())\ - + "name=%s&key=%s" - PARAMS_FILE = pp + "params" - EXT_MAP = pp + "ext-map" - EXT_REDUCE = pp + "ext-reduce" + PARAMS_FILE = pp + "params.dl" + EXT_MAP = pp + "ext.map" + EXT_REDUCE = pp + "ext.reduce" PART_SUFFIX = "-%.9d" MAP_OUTPUT = pp + "map-disco-%d" + PART_SUFFIX - CHUNK_OUTPUT = pp + "map-chunk-%d" + PART_OUTPUT = pp + "part-disco-%.9d" REDUCE_DL = pp + "reduce-in-%d.dl" REDUCE_SORTED = pp + "reduce-in-%d.sorted" REDUCE_OUTPUT = pp + "reduce-disco-%d" OOB_FILE = pp + "oob/%s" + MAP_INDEX = pp + "map-index.txt" + REDUCE_INDEX = pp + "reduce-index.txt" def put(key, value): if oob_chars.match(key): @@ -96,45 +99,33 @@ def put(key, value): def get(key, job = None): try: - if job: - url = OOB_URL % (job, key) - else: - url = OOB_URL % (this_name(), key) - return comm.download(url, redir = True) + job = job or this_name() + return load_oob("http://" + this_master(), job, key) except comm.CommException, x: data_err("OOB key (%s) not found at %s: HTTP status '%s'" %\ (key, url, x.http_code), key) -def open_local(input, fname, is_chunk): +def open_local(input, fname): try: f = file(fname) - if is_chunk: - f.seek(this_partition() * 8) - start, end = struct.unpack("QQ", f.read(16)) - sze = end - start - f.seek(start) - else: - sze = os.stat(fname).st_size + sze = os.stat(fname).st_size return sze, f except: data_err("Can't access a local input file (%s): %s"\ % (input, fname), input) -def open_remote(input, ext_host, ext_file, is_chunk): +def open_remote(input, ext_host, ext_file): try: - return comm.open_remote("http://%s%s" % (ext_host, ext_file), - this_partition(), is_chunk) + return comm.open_remote("http://%s%s" % (ext_host, ext_file)) except Exception, x: data_err("Can't access an external input file (%s/%s): %s"\ % (ext_host, ext_file, x), x) def connect_input(input): - - is_chunk = input.startswith("chunk://") - - if input.startswith("disco://") or is_chunk: + + if input.startswith("disco://"): host, fname = input[8:].split("/", 1) - local_file = LOCAL_PATH + fname + local_file = "%s/data/%s" % (DISCO_ROOT, fname) ext_host = "%s:%s" % (host, HTTP_PORT) ext_file = "/" + fname @@ -146,24 +137,20 @@ def connect_input(input): elif input.startswith("raw://"): return len(input) - 6, cStringIO.StringIO(input[6:]) - else: host = this_host() - if input.startswith("chunkfile://"): - is_chunk = True - local_file = LOCAL_PATH + input[12:] - elif input.startswith("dfs://"): + if input.startswith("dfs://"): t, path = input[6:].split("/", 1) - local_file = "%s/input/%s" % (LOCAL_PATH[:-6], path) + local_file = "%s/input/%s" % (DISCO_ROOT, path) elif input.startswith("file://"): local_file = input[7:] else: local_file = input - if host == this_host() and local_file: - return open_local(input, local_file, is_chunk) + if local_file and (resultfs_enabled or host == this_host()): + return open_local(input, local_file) else: - return open_remote(input, ext_host, ext_file, is_chunk) + return open_remote(input, ext_host, ext_file) class MapOutput: def __init__(self, part, params, combiner = None): @@ -227,9 +214,12 @@ class ReduceReader: part = PART_SUFFIX % this_partition() for input in input_files: if input.startswith("dir://"): - self.inputs += [x for x in parse_dir(input)\ - if x.startswith("chunk://") or\ - x.endswith(part)] + try: + self.inputs += parse_dir(input, + part_id = this_partition()) + except: + data_err("Couldn't resolved address %s"\ + % input, input) else: self.inputs.append(input) @@ -336,23 +326,9 @@ def run_map(job_input, partitions, param): msg("Done: %d entries mapped in total" % i) -def merge_chunks(partitions): - mapout = CHUNK_OUTPUT % this_partition() - - f = file(mapout + ".partial", "w") - offset = (len(partitions) + 1) * 8 - for p in partitions: - f.write(struct.pack("Q", offset)) - offset += os.stat(p.fname).st_size - f.write(struct.pack("Q", offset)) - f.close() - - if subprocess.call("cat %s >> %s.partial" % - (" ".join([p.fname for p in partitions]), - mapout), shell = True): - data_err("Couldn't create a chunk", mapout) - os.rename(mapout + ".partial", mapout) - for p in partitions: +def merge_partitions(partitions): + for i, p in enumerate(partitions): + safe_append(file(p.fname), PART_OUTPUT % i) os.remove(p.fname) def import_modules(modules, funcs): @@ -370,6 +346,7 @@ def op_map(job): " ".join(job_input)) nr_reduces = int(job['nr_reduces']) + nr_part = max(1, nr_reduces) fun_map_reader.func_code = marshal.loads(job['map_reader']) fun_map_writer.func_code = marshal.loads(job['map_writer']) fun_partition.func_code = marshal.loads(job['partition']) @@ -396,18 +373,27 @@ def op_map(job): if 'combiner' in job: fun_combiner.func_code = marshal.loads(job['combiner']) partitions = [MapOutput(i, map_params, fun_combiner)\ - for i in range(nr_reduces)] + for i in range(nr_part)] else: - partitions = [MapOutput(i, map_params) for i in range(nr_reduces)] + partitions = [MapOutput(i, map_params) for i in range(nr_part)] run_map(job_input[0], partitions, map_params) + external.close_ext() + for p in partitions: p.close() - if 'chunked' in job: - merge_chunks(partitions) - - external.close_ext() - msg("dir://%s/map/%s" % (this_host(), JOB_HOME), "OUT") + + if nr_reduces: + merge_partitions(partitions) + n = os.path.basename(PART_OUTPUT % 0) + msg("dir://%s/%s%s:%d" % (this_host(), JOB_HOME, n, + len(partitions) - 1), "OUT") + else: + res = [os.path.basename(p.fname) for p in partitions] + index = cStringIO.StringIO("\n".join(res) + "\n") + safe_append(index, MAP_INDEX) + msg("dir://%s/%smap-index.txt" %\ + (this_host(), JOB_HOME), "OUT") def op_reduce(job): job_inputs = this_inputs() @@ -447,7 +433,9 @@ def op_reduce(job): red_out.close() external.close_ext() - - msg("dir://%s/reduce/%s" % (this_host(), JOB_HOME), "OUT") + + index = cStringIO.StringIO(os.path.basename(red_out.fname) + "\n") + safe_append(index, REDUCE_INDEX) + msg("dir://%s/%sreduce-index.txt" % (this_host(), JOB_HOME), "OUT") init() node/disconode/disco_worker.py @@ -2,11 +2,7 @@ import os, os.path, time, struct, marshal from subprocess import * from disco.netstring import decode_netstring_str from disconode.util import * - -try: - import disco.comm_curl as comm -except: - import disco.comm_httplib as comm +from disco.util import msg MAX_ITEM_SIZE = 1024**3 MAX_NUM_OUTPUT = 1000000 @@ -79,6 +75,7 @@ def prepare(ext_job, params, path): open_ext(path + "/op", params) def open_ext(fname, params): + # XXX! Run external programs in /data/ dir, not /temp/ global proc, in_fd, out_fd proc = Popen([fname], stdin = PIPE, stdout = PIPE) in_fd = proc.stdin @@ -94,28 +91,4 @@ def write_files(ext_data, path): for fname, data in ext_data.iteritems(): ensure_file(path + "/" + fname, data = data) -def ensure_file(fname, data = None, url = None, timeout = 60, mode = 500): - while timeout > 0: - if os.path.exists(fname): - return - try: - fd = os.open(fname + ".partial", - os.O_CREAT | os.O_EXCL | os.O_WRONLY, 500) - if data: - os.write(fd, data) - else: - os.write(fd, comm.download(url)) - os.close(fd) - os.rename(fname + ".partial", fname) - return - except OSError, x: - # File exists - if x.errno == 17: - time.sleep(1) - timeout -= 1 - else: - msg("Writing external file %s failed" % fname) - raise - msg("Timeout in writing external file %s" % fname) - raise Exception("Timeout in writing external file %s" % fname) node/disconode/external.py @@ -1,6 +1,5 @@ -import sys, time, os, traceback - -job_name = "none" +import sys, time, os, traceback, fcntl +from disco.util import msg, data_err, err def ensure_path(path, check_exists = True): if check_exists and os.path.exists(path): @@ -18,3 +17,68 @@ def ensure_path(path, check_exists = True): pass else: raise x + +# About concurrent append operations: +# +# Posix spec says: +# +# If the O_APPEND flag of the file status flags is set, the file +# offset shall be set to the end of the file prior to each write and no +# intervening file modification operation shall occur between changing the +# file offset and the write operation. +# +# See also +# http://www.perlmonks.org/?node_id=486488 +# +def safe_append(instream, outfile, timeout = 60): + outstream = file(outfile, "a") + while timeout > 0: + try: + fcntl.flock(outstream, fcntl.LOCK_EX | fcntl.LOCK_NB) + try: + while True: + buf = instream.read(8192) + if not buf: + instream.close() + outstream.close() + return + outstream.write(buf) + except Exception, x: + # output file is inconsistent state + # we must crash the job + err("Updating file %s failed: %s" %\ + (outfile, x)) + except IOError, x: + # Python doc guides us to check both the + # EWOULDBLOCK (11) and EACCES (13) errors + if x.errno == 11 or x.errno == 13: + time.sleep(1) + timeout -= 1 + else: + raise + data_err("Timeout when updating file %s" % outfile, outfile) + + +def ensure_file(fname, data = None, timeout = 60, mode = 500): + while timeout > 0: + if os.path.exists(fname): + return False + try: + fd = os.open(fname + ".partial", + os.O_CREAT | os.O_EXCL | os.O_WRONLY, mode) + if type(data) == str: + os.write(fd, data) + else: + os.write(fd, data()) + os.close(fd) + os.rename(fname + ".partial", fname) + return True + except OSError, x: + # File exists + if x.errno == 17: + time.sleep(1) + timeout -= 1 + else: + data_err("Writing external file %s failed"\ + % fname, fname) + data_err("Timeout in writing external file %s" % fname, fname) node/disconode/util.py @@ -2,9 +2,6 @@ import cStringIO, struct, time, sys, os from pycurl import * from disco.comm_httplib import CommException -if "nocurl" in os.environ.get("DISCO_FLAGS", "").lower().split(): - raise Exception("nocurl") - MAX_BUF = 1024**2 MAX_RETRIES = 10 dl_handle = None @@ -45,14 +42,14 @@ def download(url, data = None, redir = False): return outbuf.getvalue() class CurlConn: - def __init__(self, url, start = None, end = None, handle = None): + def __init__(self, url, handle = None, expect = 200): if handle: self.handle = handle else: self.handle = Curl() for i in range(MAX_RETRIES): - self.init_handle(url, start, end) + self.init_handle(url) self.perform() x, succ, fail = self.multi.info_read(1) if not fail: @@ -71,18 +68,12 @@ class CurlConn: code = self.handle.getinfo(HTTP_CODE) if code == 0: raise CommException("Couldn't receive http response") - - if start == None: - check_code(self.handle, 200) - else: - check_code(self.handle, 206) + check_code(self.handle, expect) - def init_handle(self, url, start, end): + def init_handle(self, url): self.handle.setopt(URL, url) self.handle.setopt(WRITEFUNCTION, self.write) self.handle.setopt(HEADERFUNCTION, self.head) - if start != None: - self.handle.setopt(RANGE, "%d-%d" % (start, end - 1)) self.multi = CurlMulti() self.multi.add_handle(self.handle) self.buf = "" @@ -102,8 +93,15 @@ class CurlConn: self.cont = num_handles def head(self, buf): - if buf.lower().startswith("content-length:"): + buf = buf.lower() + if buf.startswith("content-length:"): self.length = int(buf.split(":")[1]) + elif buf.startswith("location:"): + self.location = buf.split(":", 1)[1].strip() + + def getheader(self, x): + if x == "location": + return self.location def write(self, buf): self.body = True @@ -119,23 +117,9 @@ class CurlConn: def disco_stats(self): pass -def open_remote(url, part = None, is_chunk = None): +def open_remote(url, expect = 200): c = Curl() - if is_chunk: - pos = part * 8 - buf = cStringIO.StringIO() - c.setopt(URL, url) - c.setopt(RANGE, "%d-%d" % (pos, pos + 15)) - c.setopt(WRITEFUNCTION, buf.write) - c.perform() - check_code(c, 206) - start, end = struct.unpack("QQ", buf.getvalue()) - if start == end: - return 0, cStringIO.StringIO() - else: - start = end = None - - conn = CurlConn(url, start, end, handle = c) + conn = CurlConn(url, handle = c, expect = expect) return conn.length, conn pydisco/disco/comm_curl.py @@ -33,7 +33,7 @@ def check_code(fd, expected): if fd.status != expected: raise CommException(fd.status) -def open_remote(url, part = None, is_chunk = False, data = None): +def open_remote(url, data = None, expect = 200): try: ext_host, ext_file = url[7:].split("/", 1) ext_file = "/" + ext_file @@ -51,30 +51,14 @@ def open_remote(url, part = None, is_chunk = False, data = None): http = httplib.HTTPConnection(ext_host) http_pool[ext_host] = http - if is_chunk: - pos = part * 8 - rge = "bytes=%d-%d" % (pos, pos + 15) - http.request("GET", ext_file, None, {"Range": rge}) - fd = http.getresponse() - check_code(fd, 206) - - start, end = struct.unpack("QQ", fd.read()) - if start == end: - return 0, cStringIO.StringIO() - else: - rge = "bytes=%d-%d" % (start, end - 1) - - http.request("GET", ext_file, None, {"Range": rge}) - fd = http.getresponse() - check_code(fd, 206) - elif data: + if data: http.request("POST", ext_file, data) fd = http.getresponse() - check_code(fd, 200) + check_code(fd, expect) else: http.request("GET", ext_file, None) fd = http.getresponse() - check_code(fd, 200) + check_code(fd, expect) sze = fd.getheader("content-length") if sze: @@ -88,4 +72,4 @@ def open_remote(url, part = None, is_chunk = False, data = None): # endless recursion if something went seriously wrong. http.close() del http_pool[ext_host] - return open_remote(url, part, is_chunk) + return open_remote(url, data) pydisco/disco/comm_httplib.py @@ -1,12 +1,7 @@ import sys, re, os, marshal, cjson, time, cPickle -from disco import func, util +from disco import func, util, comm from netstring import * -try: - import disco.comm_curl as comm -except: - import disco.comm_httplib as comm - class JobException(Exception): def __init__(self, msg, master, name): self.msg = msg @@ -73,13 +68,11 @@ class Disco(object): def oob_get(self, name, key): try: - r = self.request("/disco/ctrl/oob_get?name=%s&key=%s" %\ - (name, key), redir = True) + return util.load_oob(self.host, name, key) except comm.CommException, x: if x.http_code == 404: raise KeyError("Unknown key or job name") raise - return r def oob_list(self, name): try: @@ -197,7 +190,6 @@ class Job(object): "sort": False, "params": Params(), "mem_sort_limit": 256 * 1024**2, - "chunked": None, "ext_params": None, "status_interval": 100000, "required_modules": [], @@ -234,6 +226,9 @@ class Job(object): if "input_files" in kw: kw["input"] = kw["input_files"] + + if "chunked" in kw: + raise Exception("Argument 'chunked' is deprecated") if not "input" in kw: raise Exception("input is required") @@ -289,24 +284,46 @@ class Job(object): parsed_inputs.append(inp) inputs = parsed_inputs else: - addr = [] + nr_maps = 0 + ext_inputs = [] + red_inputs = [] for inp in inputs: if type(inp) == list: raise Exception("Reduce doesn't "\ "accept redundant inputs") - elif not inp.startswith("dir://"): - addr.append(inp) - - if d("nr_reduces") == None and not addr: - raise Exception("nr_reduces must match to "\ - "the number of partitions in the "\ - "input data") - - if d("nr_reduces") != 1 and addr: + elif inp.startswith("dir://"): + if inp.endswith(".txt"): + ext_inputs.append(inp) + else: + red_inputs.append(inp) + else: + ext_inputs.append(inp) + + if ext_inputs and red_inputs: + raise Exception("Can't mix partitioned "\ + "inputs with other inputs") + elif red_inputs: + q = lambda x: int(x.split(":")[-1]) + 1 + nr_red = q(red_inputs[0]) + for x in red_inputs: + if q(x) != nr_red: + raise Exception(\ + "Number of partitions must "\ + "match in all inputs") + n = d("nr_reduces") or nr_red + if n != nr_red: + raise Exception( + "Specified nr_reduces = %d but "\ + "number of partitions in the input "\ + "is %d" % (n, nr_red)) + kw["nr_reduces"] = nr_red + inputs = red_inputs + elif d("nr_reduces") != 1: raise Exception("nr_reduces must be 1 when "\ - "using external inputs without "\ - "the map phase") - nr_maps = 0 + "using non-partitioned inputs "\ + "without the map phase") + else: + inputs = ext_inputs req["input"] = " ".join(inputs) req["nr_maps"] = str(nr_maps) @@ -327,7 +344,6 @@ class Job(object): req["reduce"] = marshal.dumps( kw["reduce"].func_code) nr_reduces = nr_reduces or min(max(nr_maps / 2, 1), 100) - req["chunked"] = "True" req["reduce_reader"] =\ marshal.dumps(d("reduce_reader").func_code) @@ -338,16 +354,10 @@ class Job(object): req["reduce_init"] = marshal.dumps(\ kw["reduce_init"].func_code) else: - nr_reduces = nr_reduces or 1 - + nr_reduces = nr_reduces or 0 + req["nr_reduces"] = str(nr_reduces) - if d("chunked") != None: - if d("chunked"): - req["chunked"] = "True" - elif "chunked" in req: - del req["chunked"] - if "combiner" in kw: req["combiner"] =\ marshal.dumps(kw["combiner"].func_code) @@ -377,23 +387,33 @@ def result_iterator(results, notifier = None,\ else: res.append(dir_url) + x, x, root = util.load_conf() + for url in res: if url.startswith("file://"): fname = url[7:] fd = file(fname) sze = os.stat(fname).st_size - else: + elif url.startswith("disco://"): host, fname = url[8:].split("/", 1) if proxy: ext_host = proxy fname = "/disco/node/%s/%s" % (host, fname) else: ext_host = host + ":" + util.HTTP_PORT - sze, fd = comm.open_remote("http://%s/%s" % (ext_host, fname)) + if util.resultfs_enabled: + f = "%s/data/%s" % (root, fname) + fd = file(f) + sze = os.stat(f).st_size + else: + sze, fd = comm.open_remote("http://%s/%s"\ + % (ext_host, fname)) + else: + raise JobException("Invalid result url: %s" % url) if notifier: notifier(url) - for x in reader(fd, fd.length, fname): + for x in reader(fd, sze, fname): yield x pydisco/disco/core.py @@ -1,13 +1,11 @@ import re, os import sys, time, os, traceback - -try: - import disco.comm_curl as comm -except: - import disco.comm_httplib as comm +from disco import comm job_name = "none" +resultfs_enabled =\ + "resultfs" in os.environ.get("DISCO_FLAGS", "").lower().split() def msg(m, c = 'MSG', job_input = ""): t = time.strftime("%y/%m/%d %H:%M:%S") @@ -35,14 +33,14 @@ def load_conf(): return os.environ.get("DISCO_MASTER_PORT", master.strip()),\ os.environ.get("DISCO_PORT", port.strip()),\ - os.environ.get("DISCO_ROOT", root.strip()) + "/data/" + os.environ.get("DISCO_ROOT", root.strip()) def jobname(addr): if addr.startswith("disco:") or addr.startswith("http:"): return addr.strip("/").split("/")[-2] elif addr.startswith("dir:"): - return addr.strip("/").split("/")[-1] + return addr.strip("/").split("/")[-2] else: raise "Unknown address: %s" % addr @@ -70,15 +68,45 @@ def disco_host(addr): raise "Unknown host specifier: %s" % addr -def parse_dir(dir_url, proxy = None): - x, x, host, mode, name = dir_url.split("/", 4) +def parse_dir(dir_url, proxy = None, part_id = None): + x, x, host, name = dir_url.split("/", 3) if proxy: - url = "http://%s/disco/node/%s/%s/" % (proxy, host, name) + url = "http://%s/disco/node/%s/%s" % (proxy, host, name) + else: + url = "http://%s:%s/%s" % (host, HTTP_PORT, name) + + if name.endswith(".txt"): + if resultfs_enabled: + r = file("%s/data/%s" % (ROOT, name)).readlines() + else: + r = comm.download(url).splitlines() + else: + b, max = name.split("/")[-1].split(":") + fl = len(max) + base = b[:len(b) - fl] + t = "%s%%.%dd" % (base, fl) + if part_id != None: + r = [t % part_id] + else: + r = [t % i for i in range(int(max) + 1)] + + p = "/".join(name.split("/")[:-1]) + return ["disco://%s/%s/%s" % (host, p, x.strip()) for x in r] + +def load_oob(host, name, key): + url = "%s/disco/ctrl/oob_get?name=%s&key=%s" % (host, name, key) + if resultfs_enabled: + sze, fd = comm.open_remote(url, expect = 302) + loc = fd.getheader("location") + fname = "%s/data/%s" % (ROOT, "/".join(loc.split("/")[3:])) + try: + return file(fname).read() + except Exception, x: + raise comm.CommException(404, + "OOB key (%s) not found at %s" %\ + (key, fname)) else: - url = "http://%s:%s/%s/" % (host, HTTP_PORT, name) - html = comm.download(url) - inputs = re.findall(">(%s-(.+?)-.*?)</a>" % mode, html) - return ["%s://%s/%s/%s" % (prefix, host, name, x)\ - for x, prefix in inputs if "." not in x] + return comm.download(url, redir = True) + -MASTER_PORT, HTTP_PORT, tmp = load_conf() +MASTER_PORT, HTTP_PORT, ROOT = load_conf() pydisco/disco/util.py @@ -30,16 +30,17 @@ DEFAULT_NUFA_PORT = 9800 def check_config(config, replicas = True): REQ = ["nodes", "volumes", "master", "config_dir"] if replicas: - REQ.append("replicas") + REQ.append("replicas") + if config["replicas"] > len(config["nodes"]): + print "replicas must be less than equal"\ + "to the number of nodes." + print "Check the config file." + sys.exit(1) for k in REQ: if k not in config: print "Required field '%s' is missing." % k print "Check the config file." sys.exit(1) - if config["replicas"] > len(config["nodes"]): - print "replicas must be less than equal to the number of nodes." - print "Check the config file." - sys.exit(1) def output_volume(f, name, type, subvol = None, options = {}): print >> f, "volume %s" % name @@ -54,7 +55,8 @@ def output_volume(f, name, type, subvol = None, options = {}): def nodes_sect(f, config): print >> f, "\n# -----\n# NODES\n# -----\n" opt = {"transport-type": "tcp", - "remote-port": config["port"]} + "remote-port": config["port"], + "ping-timeout": "60"} for node in config["nodes"]: print >> f, "# -- %s\n" % node @@ -105,12 +107,16 @@ def server_sect(f, config, options, writevol): output_volume(f, "%s-posix" % vol, "storage/posix", [], {"directory": dir}) subvol.append(vol) - output_volume(f, "%s-lock" % vol, "features/locks", + output_volume(f, "%s-lock" % vol, + "features/locks", ["%s-posix" % vol]) - output_volume(f, vol, "performance/read-ahead", + output_volume(f, "%s-readahead" % vol, + "performance/read-ahead", ["%s-lock" % vol], { - "page-size": "512KB", - "page-count": "4"}) + "page-size": "512KB", + "page-count": "4"}) + output_volume(f, vol, "performance/io-threads", + ["%s-readahead" % vol], {"thread-count": "32"}) else: print >> f, "# this is just a dummy volume as gluster" print >> f, "# requires us to specify a subvolume for" @@ -118,7 +124,7 @@ def server_sect(f, config, options, writevol): print >> f, "# here." path = os.path.abspath(config["config_dir"]) output_volume(f, "masterdummy", "storage/posix", - [], {"directory": path}) + [], {"directory": "/tmp"}) subvol = ["masterdummy"] print >> f, "# -- server\n" @@ -148,6 +154,10 @@ def create_nufa_config(config, path): if len(config["volumes"]) > 1: print "Specify only one volume for results" sys.exit(1) + if config["master"] not in config["nodes"]: + print "Add your master node '%s' to the list of nodes" %\ + config["master"] + sys.exit(1) config["port"] = config.get("port", DEFAULT_NUFA_PORT) util/gluster_config.py test/test_chunked.py e84effe473b7935d0766c484732c102c70e191c2 Ville Tuulos tuulos@nxfront.nokiapaloalto.com http://github.com/tuulos/disco/commit/2ae75c30afee7a42d1bd73eb66f62334f44b104d 2ae75c30afee7a42d1bd73eb66f62334f44b104d 2009-06-16T19:57:51-07:00 2009-06-11T15:27:13-07:00 Support for resultfs finalized, new disk layout for intermediate results. Closes #9 291a8102874ed5c50589598823505b80ee738874 Ville Tuulos tuulos@nxfront.nokiapaloalto.com