pydisco/disco/error.py @@ -35,7 +35,7 @@ init(_Args) -> % disco_worker processes. ets:new(active_workers, [named_table, public]), - % node_laod records how many disco_workers there are + % node_load records how many disco_workers are % running on a node (could be found in active_workers % as well). This table exists mainly for convenience and % possibly for performance reasons. master/src/disco_server.erl @@ -3,8 +3,7 @@ import re, traceback, tempfile, struct, random 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.util import ensure_path, safe_append, write_files from disconode import external from disco import comm @@ -160,7 +159,7 @@ def connect_input(input): else: return open_remote(input, ext_host, ext_file) -class MapOutput: +class MapOutput(object): def __init__(self, part, params, combiner = None): self.combiner = combiner self.params = params @@ -193,7 +192,7 @@ class MapOutput: os.rename(self.fname + ".partial", self.fname) -class ReduceOutput: +class ReduceOutput(object): def __init__(self, params): self.fname = REDUCE_OUTPUT % this_partition() self.params = params @@ -216,7 +215,7 @@ def num_cmp(x, y): pass return cmp(x, y) -class ReduceReader: +class ReduceReader(object): def __init__(self, input_files, do_sort, mem_sort_limit): self.inputs = [] part = PART_SUFFIX % this_partition() node/disconode/disco_worker.py @@ -1,8 +1,9 @@ import os, os.path, time, struct, marshal -from subprocess import * +from subprocess import Popen, PIPE from disco.netstring import decode_netstring_str -from disconode.util import * +from disconode.util import write_files from disco.util import msg +from disco.error import DiscoError MAX_ITEM_SIZE = 1024**3 MAX_NUM_OUTPUT = 1000000 @@ -16,11 +17,11 @@ def pack_kv(k, v): def unpack_kv(): le = struct.unpack("I", out_fd.read(4))[0] if le > MAX_ITEM_SIZE: - raise "External key size exceeded: %d bytes" % le + raise DiscoError("External key size exceeded: %d bytes" % le) k = out_fd.read(le) le = struct.unpack("I", out_fd.read(4))[0] if le > MAX_ITEM_SIZE: - raise "External key size exceeded: %d bytes" % le + raise DiscoError("External key size exceeded: %d bytes" % le) v = out_fd.read(le) return k, v @@ -48,14 +49,14 @@ def ext_reduce(red_in, red_out, params): while True: for fd, event in p.poll(): if event & (select.POLLNVAL | select.POLLERR): - raise "Pipe to the external process failed" + raise DiscoError("Pipe to the external process failed") elif event & select.POLLIN: num = struct.unpack("I", external.out_fd.read(4))[0] if num > MAX_NUM_OUTPUT: - raise "External output limit "\ + raise DiscoError("External output limit "\ "exceeded: %d > %d" %\ - (num, MAX_NUM_OUTPUT) + (num, MAX_NUM_OUTPUT)) for i in range(num): red_out.add(*external.unpack_kv()) tt += 1 node/disconode/external.py @@ -6,9 +6,9 @@ def ensure_path(path, check_exists = True): err("File exists: %s" % path) if os.path.isfile(path): os.remove(path) - dir, fname = os.path.split(path) + dirpath, fname = os.path.split(path) try: - os.makedirs(dir) + os.makedirs(dirpath) except OSError, x: if x.errno == 17: # File exists is ok, it may happen node/disconode/util.py @@ -1,24 +0,0 @@ -import sys -from types import FunctionType -from core import Disco, Params, result_iterator -import func - -class DeprecatedFunc: - def __init__(self, fun, name): - self.fun = fun - self.name = name - - def __getattr__(self, name): - if name == "func_code": - print >> sys.stderr, "WARNING! disco.%s is "\ - "deprecated. Use disco.func.%s instead."\ - % (self.name, self.name) - return self.fun.func_code - raise AttributeException("%s not found" % name) - -chain_reader = DeprecatedFunc(func.chain_reader, "chain_reader") - -def job(master, name, input, fun_map = None, **kw): - kw.update({"name": name, "input": input}) - kw["map"] = kw.get("map", fun_map) - return Disco(master).new_job(**kw).wait(clean = kw.get("clean", True)) pydisco/disco/__init__.py @@ -4,11 +4,24 @@ nocurl = "nocurl" in os.environ.get("DISCO_FLAGS", "").lower().split() try: import pycurl -except: +except ImportError: nocurl = True if nocurl: - from comm_httplib import * + from disco.comm_httplib import * else: - from comm_curl import * + from disco.comm_curl import * +# get rid of this for python2.6+ +try: + import json +except ImportError: + try: + import simplejson as json + except ImportError: + import cjson + class Dummy(object): + pass + json = Dummy() + json.loads = cjson.decode + json.dumps = cjson.encode pydisco/disco/comm.py @@ -1,4 +1,4 @@ -import httplib, urllib2, cStringIO, struct +import httplib, urllib http_pool = {} pydisco/disco/comm_httplib.py @@ -1,19 +1,10 @@ -import sys, re, os, marshal, cjson, time, cPickle, types, cStringIO -from disco import func, util, comm, modutil, eventmonitor -from netstring import * - -class JobException(Exception): - def __init__(self, msg, master, name): - self.msg = msg - self.name = name - self.master = master - - def __str__(self): - return "Job %s/%s failed: %s" %\ - (self.master, self.name, self.msg) - +import sys, re, os, marshal, time, cPickle +from disco import func, util, comm +from disco.comm import json +from disco.error import DiscoError, JobException +from disco.netstring import encode_netstring_fd, decode_netstring_fd -class Params: +class Params(object): def __init__(self, **kwargs): self._state = {} for k, v in kwargs.iteritems(): @@ -56,22 +47,24 @@ class Disco(object): def __init__(self, host): self.host = "http://" + util.disco_host(host)[7:] - def request(self, url, data = None, redir = False, offset = 0): - return comm.download(self.host + url,\ - data = data, redir = redir, offset = offset) - + def request(self, url, data = None, redir = False): + try: + return comm.download(self.host + url, data = data, redir = redir) + except comm.CommException, msg: + raise DiscoError("Failed to fetch %r, data of len %r, redir=%r. Error was %s" % (self.host + url, len(data), redir, msg)) + def nodeinfo(self): - return cjson.decode(self.request("/disco/ctrl/nodeinfo")) + return json.loads(self.request("/disco/ctrl/nodeinfo")) def joblist(self): - return cjson.decode(self.request("/disco/ctrl/joblist")) + return json.loads(self.request("/disco/ctrl/joblist")) def oob_get(self, name, key): try: 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 DiscoError("Unknown key or job name") raise def oob_list(self, name): @@ -80,9 +73,9 @@ class Disco(object): redir = True) except comm.CommException, x: if x.http_code == 404: - raise KeyError("Unknown key or job name") + raise DiscoError("Unknown key or job name") raise - return cjson.decode(r) + return json.loads(r) def profile_stats(self, name, mode = ""): import pstats @@ -125,8 +118,8 @@ class Disco(object): if not len(l): continue try: - ent = tuple(cjson.decode(l)) - except cjson.DecodeError: + ent = tuple(json.loads(l)) + except ValueError: break # HTTP range request doesn't like empty ranges: # Let's ensure that at least the last newline @@ -158,10 +151,10 @@ class Disco(object): nam.append(n.name) r = self.request("/disco/ctrl/get_results", - cjson.encode([timeout, nam])) + json.dumps([timeout, nam])) if not r: return None - r = cjson.decode(r) + r = json.loads(r) if single: return r[0][1] else: @@ -177,7 +170,7 @@ class Disco(object): def jobinfo(self, name): r = self.request("/disco/ctrl/jobinfo?name=" + name) if r: - return cjson.decode(r) + return json.loads(r) else: return r @@ -244,9 +237,9 @@ class Job(object): def __init__(self, master, **kwargs): self.master = master if "name" not in kwargs: - raise Exception("Argument name is required") + raise DiscoError("Argument name is required") if re.search("\W", kwargs["name"]): - raise Exception("Only characters in [a-zA-Z0-9_] "\ + raise DiscoError("Only characters in [a-zA-Z0-9_] "\ "are allowed in the job name") self.name = "%s@%d" % (kwargs["name"], int(time.time())) self._run(**kwargs) @@ -276,17 +269,17 @@ class Job(object): kw["input"] = kw["input_files"] if "chunked" in kw: - raise Exception("Argument 'chunked' is deprecated") + raise DiscoError("Argument 'chunked' is deprecated") if not "input" in kw: - raise Exception("input is required") + raise DiscoError("input is required") if not ("map" in kw or "reduce" in kw): - raise Exception("Specify map and/or reduce") + raise DiscoError("Specify map and/or reduce") for p in kw: if p not in Job.defaults: - raise Exception("Unknown argument: %s" % p) + raise DiscoError("Unknown argument: %s" % p) inputs = kw["input"] @@ -382,7 +375,7 @@ class Job(object): red_inputs = [] for inp in inputs: if type(inp) == list: - raise Exception("Reduce doesn't "\ + raise DiscoError("Reduce doesn't "\ "accept redundant inputs") elif inp.startswith("dir://"): if inp.endswith(".txt"): @@ -393,26 +386,26 @@ class Job(object): ext_inputs.append(inp) if ext_inputs and red_inputs: - raise Exception("Can't mix partitioned "\ + raise DiscoError("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(\ + raise DiscoError(\ "Number of partitions must "\ "match in all inputs") n = d("nr_reduces") or nr_red if n != nr_red: - raise Exception( + raise DiscoError( "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 "\ + raise DiscoError("nr_reduces must be 1 when "\ "using non-partitioned inputs "\ "without the map phase") else: @@ -459,7 +452,7 @@ class Job(object): reply = self.master.request("/disco/job/new", self.msg) if reply != "job started": - raise Exception("Failed to start a job. Server replied: " + reply) + raise DiscoError("Failed to start a job. Server replied: " + reply) pydisco/disco/core.py @@ -1,5 +1,7 @@ -import sys, cjson +import sys from os import environ +from disco.comm import json +from disco.error import DiscoError # Following from Python cookbook, #475186 def has_colors(stream): @@ -27,7 +29,7 @@ class EventMonitor(object): self.name = name if not disco: - raise Exception("Specify either job or disco and name") + raise DiscoError("Specify either job or disco and name") self.offset = 0 self.output = None @@ -101,7 +103,7 @@ class EventMonitor(object): def json_output(self, status, tstamp, host, msg): if tstamp: - print cjson.encode([tstamp, host, msg]) + print json.dumps([tstamp, host, msg]) def ansi_output(self, status, tstamp, host, msg): if status: pydisco/disco/eventmonitor.py @@ -2,6 +2,7 @@ import re, cPickle from disco.util import err, data_err, msg def netstr_reader(fd, content_len, fname): + if content_len == None: err("Content-length must be defined for netstr_reader") def read_netstr(idx, data, tot): pydisco/disco/func.py @@ -24,11 +24,14 @@ import StringIO MAX_LEN_STRING = 10 MAX_PACKET_LEN = 1024**3 +class NetStringError(Exception): + pass + def _read_string(msg, i): j = msg.index(" ", i) - len = int(msg[i: j]) + length = int(msg[i: j]) j += 1 - return (j + len + 1, msg[j: j + len]) + return (j + length + 1, msg[j: j + length]) def encode_netstring_str(d): @@ -66,13 +69,13 @@ def decode_netstring_fd(fd): lenstr += c i += 1 if i > MAX_LEN_STRING: - raise "Length string too long" + raise NetStringError("Length string too long") if not lenstr: raise EOFError() - llen = int(lenstr) - if llen > MAX_PACKET_LEN: - raise "Will not receive %d bytes" % llen + length = int(lenstr) + if length > MAX_PACKET_LEN: + raise NetStringError("Will not receive %d bytes" % length) - return decode_netstring_str(fd.read(llen)) + return decode_netstring_str(fd.read(length)) pydisco/disco/netstring.py @@ -1,7 +1,7 @@ - -import re, os -import sys, time, os, traceback -from disco import comm +import os +import sys, time, traceback +from disco.comm import CommException, download, open_remote +from disco.error import DiscoError job_name = "none" resultfs_enabled =\ @@ -14,12 +14,12 @@ def msg(m, c = 'MSG', job_input = ""): def err(m): msg(m, 'MSG') - raise Exception(m) + raise DiscoError(m) def data_err(m, job_input): msg(m, 'DAT', job_input) if sys.exc_info() == (None, None, None): - raise Exception(m) + raise DiscoError(m) else: print traceback.print_exc() raise @@ -42,7 +42,7 @@ def jobname(addr): elif addr.startswith("dir:"): return addr.strip("/").split("/")[-2] else: - raise "Unknown address: %s" % addr + raise DiscoError("Unknown address: %s" % addr) def pack_files(files): msg = {} @@ -68,7 +68,7 @@ def disco_host(addr): elif addr.startswith("http:"): return addr else: - raise "Unknown host specifier: %s" % addr + raise DiscoError("Unknown host specifier: %s" % addr) def proxy_url(proxy, path, node = "x"): if not proxy: @@ -80,7 +80,7 @@ def proxy_url(proxy, path, node = "x"): elif proxy.startswith("http://"): host = proxy[7:] else: - raise Exception("Unknown proxy protocol: %s" % proxy) + raise DiscoError("Unknown proxy protocol: %s" % proxy) return "http://%s/disco/node/%s/%s" % (host, node, path) def parse_dir(dir_url, proxy = None, part_id = None): @@ -92,14 +92,14 @@ def parse_dir(dir_url, proxy = None, part_id = None): else: r = comm.download(url).splitlines() else: - b, max = name.split("/")[-1].split(":") - fl = len(max) + b, mmax = name.split("/")[-1].split(":") + fl = len(mmax) 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)] + r = [t % i for i in range(int(mmax) + 1)] p = "/".join(name.split("/")[:-1]) return ["disco://%s/%s/%s" % (host, p, x.strip()) for x in r] @@ -114,7 +114,7 @@ def load_oob(host, name, key): fname = "%s/data/%s" % (ROOT, "/".join(loc.split("/")[3:])) try: return file(fname).read() - except Exception, x: + except DiscoError: raise comm.CommException(404, "OOB key (%s) not found at %s" %\ (key, fname)) pydisco/disco/util.py @@ -1,5 +1,6 @@ -import sys, cjson +import sys import tserver +from disco.comm import json from disco.core import Disco, result_iterator def data_gen(path): @@ -12,12 +13,12 @@ def fun_map(e, params): return [(int(e), "")] def add_node(): - orig_config = cjson.decode( + orig_config = json.loads( disco.request("/disco/ctrl/load_config_table")) config = orig_config[:] config.append(["missingnode", "2"]) r = disco.request("/disco/ctrl/save_config_table", - cjson.encode(config)) + json.dumps(config)) if r != "\"table saved!\"": raise Exception("Couldn't add a dummy node: %s" % r) return orig_config @@ -48,7 +49,7 @@ except: raise finally: disco.request("/disco/ctrl/save_config_table", - cjson.encode(orig_config)) + json.dumps(orig_config)) print "ok" test/test_missingnode.py @@ -22,7 +22,15 @@ where - config_dir: where to store config files on the master node """ -import sys, cjson, md5, cStringIO, os +import sys, cStringIO, os + +try: + import hashlib as md5 +except ImportError: + # Hashlib is not available in Python2.4 + import md5 + +from disco.comm import json DEFAULT_REPL_PORT = 9900 DEFAULT_NUFA_PORT = 9800 @@ -193,24 +201,26 @@ def create_master_config(name, config, path, client_conf): print "ok" if len(sys.argv) < 3 or sys.argv[1] not in ["inputfs", "resultfs"]: - print "\nUsage: python gluster_config.py [inputfs|resultfs] config.json" - print - print "This script generates Disco-compatible config files for Gluster," - print "a distributed filesystem." - print - print "Two modes are available:" - print "- inputfs, which produces a Gluster volfile that is suitable for" - print " storing input data for Disco so that data is k-way replicated" - print " over nodes." - print "- resultfs, which produces a Gluster volfile for communication" - print " between Disco nodes, in place of the default HTTP-based" - print " solution." - print - print "See gluster_example.json for an example config gile. For more" - print "information, see http://discoproject.org/doc/start/dfs.html.\n" + print """ + Usage: python gluster_config.py [inputfs|resultfs] config.json + + This script generates Disco-compatible config files for Gluster, + a distributed filesystem. + + Two modes are available: + - inputfs, which produces a Gluster volfile that is suitable for + storing input data for Disco so that data is k-way replicated + over nodes. + - resultfs, which produces a Gluster volfile for communication + between Disco nodes, in place of the default HTTP-based + solution. + + See gluster_example.json for an example config gile. For more + information, see http://discoproject.org/doc/start/dfs.html. + """ sys.exit(1) -config = cjson.decode(file(sys.argv[2]).read()) +config = json.loads(file(sys.argv[2]).read()) path = os.path.abspath(config["config_dir"]) if sys.argv[1] == "inputfs": util/gluster_config.py eb347eadc2ec65f868df35480be50edb7003b91c Mistobaan <> http://github.com/tuulos/disco/commit/e05ed88f1fbca696c47fda80cc45504a33707e23 e05ed88f1fbca696c47fda80cc45504a33707e23 2009-09-05T01:21:03-07:00 2009-08-25T10:29:54-07:00 mistobaans changes + some DiscoError - some import * e2c10087cda34fc5d5ad5e688ed73ccd7f54371c Ville Tuulos tuulos@dxfront.(none)