allmydata

├── frontends

├── immutable

│   └── downloader

├── introducer

├── mutable

├── scripts

├── storage

├── test

├── util

├── web

│   └── static

│   ├── css

│   └── img

└── windows

File: allmydata/client.py

class Client(node.Node, pollmixin.PollMixin):

(…)

# these four methods are the primitives for creating filenodes and

# dirnodes. The first takes a URI and produces a filenode or (new-style)

# dirnode. The other three create brand-new filenodes/dirnodes.

def create_node_from_uri(self, write_uri, read_uri=None, deep_immutable=False, name="<unknown name>"):

# This returns synchronously.

# Note that it does *not* validate the write_uri and read_uri; instead we

# may get an opaque node if there were any problems.

return self.nodemaker.create_from_cap(write_uri, read_uri, deep_immutable=deep_immutable, name=name)

def create_dirnode(self, initial_children={}, version=None):

d = self.nodemaker.create_new_mutable_directory(initial_children, version=version)

return d

def create_immutable_dirnode(self, children, convergence=None):

return self.nodemaker.create_immutable_directory(children, convergence)

def create_mutable_file(self, contents=None, keysize=None, version=None):

return self.nodemaker.create_mutable_file(contents, keysize,

version=version)

def upload(self, uploadable):

uploader = self.getServiceNamed("uploader")

return uploader.upload(uploadable)

File: allmydata/nodemaker.py

class NodeMaker:

implements(INodeMaker)

(…)

def create_mutable_file(self, contents=None, keysize=None, version=None):

if version is None:

version = self.mutable_file_default

n = MutableFileNode(self.storage_broker, self.secret_holder,

self.default_encoding_parameters, self.history)

d = self.key_generator.generate(keysize)

d.addCallback(n.create_with_keys, contents, version=version)

d.addCallback(lambda res: n)

return d

def create_new_mutable_directory(self, initial_children={}, version=None):

# initial_children must have metadata (i.e. {} instead of None)

for (name, (node, metadata)) in initial_children.iteritems():

precondition(isinstance(metadata, dict),

"create_new_mutable_directory requires metadata to be a dict, not None", metadata)

node.raise_error()

d = self.create_mutable_file(lambda n:

MutableData(pack_children(initial_children,

n.get_writekey())),

version=version)

d.addCallback(self._create_dirnode)

return d

def create_immutable_directory(self, children, convergence=None):

if convergence is None:

convergence = self.secret_holder.get_convergence_secret()

packed = pack_children(children, None, deep_immutable=True)

uploadable = Data(packed, convergence)

d = self.uploader.upload(uploadable)

d.addCallback(lambda results:

self.create_from_cap(None, results.get_uri()))

d.addCallback(self._create_dirnode)

return d

File: allmydata/immutable/upload.py

class Uploader(service.MultiService, log.PrefixingLogMixin):

"""I am a service that allows file uploading. I am a service-child of the

Client.

"""

(...)

def upload(self, uploadable):

"""

Returns a Deferred that will fire with the UploadResults instance.

"""

assert self.parent

assert self.running

uploadable = IUploadable(uploadable)

d = uploadable.get_size()

def _got_size(size):

default_params = self.parent.get_encoding_parameters()

precondition(isinstance(default_params, dict), default_params)

precondition("max_segment_size" in default_params, default_params)

uploadable.set_default_encoding_parameters(default_params)

if self.stats_provider:

self.stats_provider.count('uploader.files_uploaded', 1)

self.stats_provider.count('uploader.bytes_uploaded', size)

if size <= self.URI_LIT_SIZE_THRESHOLD:

uploader = LiteralUploader()

return uploader.start(uploadable)

else:

eu = EncryptAnUploadable(uploadable, self._parentmsgid)

d2 = defer.succeed(None)

storage_broker = self.parent.get_storage_broker()

if self._helper:

uploader = AssistedUploader(self._helper, storage_broker)

d2.addCallback(lambda x: eu.get_storage_index())

d2.addCallback(lambda si: uploader.start(eu, si))

else:

storage_broker = self.parent.get_storage_broker()

secret_holder = self.parent._secret_holder

uploader = CHKUploader(storage_broker, secret_holder)

d2.addCallback(lambda x: uploader.start(eu))

self._all_uploads[uploader] = None

if self._history:

self._history.add_upload(uploader.get_upload_status())

def turn_verifycap_into_read_cap(uploadresults):

# Generate the uri from the verifycap plus the key.

d3 = uploadable.get_encryption_key()

def put_readcap_into_results(key):

v = uri.from_string(uploadresults.get_verifycapstr())

r = uri.CHKFileURI(key, v.uri_extension_hash, v.needed_shares, v.total_shares, v.size)

uploadresults.set_uri(r.to_string())

return uploadresults

d3.addCallback(put_readcap_into_results)

return d3

d2.addCallback(turn_verifycap_into_read_cap)

return d2

d.addCallback(_got_size)

def _done(res):

uploadable.close()

return res

d.addBoth(_done)

return d

File: allmydata/mutable/filenode.py

class MutableFileNode:

implements(IMutableFileNode, Icheckable)

implements(IMutableFileNode, ICheckable)

def __init__(self, storage_broker, secret_holder,

default_encoding_parameters, history):

self._storage_broker = storage_broker

self._secret_holder = secret_holder

self._default_encoding_parameters = default_encoding_parameters

self._history = history

self._pubkey = None # filled in upon first read

self._privkey = None # filled in if we're mutable

# we keep track of the last encoding parameters that we use. These

# are updated upon retrieve, and used by publish. If we publish

# without ever reading (i.e. overwrite()), then we use these values.

self._required_shares = default_encoding_parameters["k"]

self._total_shares = default_encoding_parameters["n"]

self._sharemap = {} # known shares, shnum-to-[nodeids]

self._most_recent_size = None

(...)

def create_with_keys(self, (pubkey, privkey), contents,

version=SDMF_VERSION):

"""Call this to create a brand-new mutable file. It will create the

shares, find homes for them, and upload the initial contents (created

with the same rules as IClient.create_mutable_file() ). Returns a

Deferred that fires (with the MutableFileNode instance you should

use) when it completes.

"""

self._pubkey, self._privkey = pubkey, privkey

pubkey_s = self._pubkey.serialize()

privkey_s = self._privkey.serialize()

self._writekey = hashutil.ssk_writekey_hash(privkey_s)

self._encprivkey = self._encrypt_privkey(self._writekey, privkey_s)

self._fingerprint = hashutil.ssk_pubkey_fingerprint_hash(pubkey_s)

if version == MDMF_VERSION:

self._uri = WriteableMDMFFileURI(self._writekey, self._fingerprint)

self._protocol_version = version

elif version == SDMF_VERSION:

self._uri = WriteableSSKFileURI(self._writekey, self._fingerprint)

self._protocol_version = version

self._readkey = self._uri.readkey

self._storage_index = self._uri.storage_index

initial_contents = self._get_initial_contents(contents)

return self._upload(initial_contents, None)

(…)

def overwrite(self, new_contents):

"""

I overwrite the contents of the best recoverable version of this

mutable file with new_contents. This is equivalent to calling

overwrite on the result of get_best_mutable_version with

new_contents as an argument. I return a Deferred that eventually

fires with the results of my replacement process.

"""

# TODO: Update downloader hints.

return self._do_serialized(self._overwrite, new_contents)

(…)

def upload(self, new_contents, servermap):

"""

I overwrite the contents of the best recoverable version of this

mutable file with new_contents, using servermap instead of

creating/updating our own servermap. I return a Deferred that

fires with the results of my upload.

"""

# TODO: Update downloader hints

return self._do_serialized(self._upload, new_contents, servermap)

def modify(self, modifier, backoffer=None):

"""

I modify the contents of the best recoverable version of this

mutable file with the modifier. This is equivalent to calling

modify on the result of get_best_mutable_version. I return a

Deferred that eventually fires with an UploadResults instance

describing this process.

"""

# TODO: Update downloader hints.

return self._do_serialized(self._modify, modifier, backoffer)

File: allmydata/client.py

class Client(node.Node, pollmixin.PollMixin):

(…)

def init_storage(self):

# should we run a storage server (and publish it for others to use)?

if not self.get_config("storage", "enabled", True, boolean=True):

return

readonly = self.get_config("storage", "readonly", False, boolean=True)

storedir = os.path.join(self.basedir, self.STOREDIR)

data = self.get_config("storage", "reserved_space", None)

try:

reserved = parse_abbreviated_size(data)

except ValueError:

log.msg("[storage]reserved_space= contains unparseable value %s"

% data)

raise

if reserved is None:

reserved = 0

(…)

ss = StorageServer(storedir, self.nodeid,

reserved_space=reserved,

discard_storage=discard,

readonly_storage=readonly,

stats_provider=self.stats_provider,

expiration_enabled=expire,

expiration_mode=mode,

expiration_override_lease_duration=o_l_d,

expiration_cutoff_date=cutoff_date,

expiration_sharetypes=expiration_sharetypes)

self.add_service(ss)

d = self.when_tub_ready()

# we can't do registerReference until the Tub is ready

def _publish(res):

furl_file = os.path.join(self.basedir, "private", "storage.furl").encode(get_filesystem_encoding())

furl = self.tub.registerReference(ss, furlFile=furl_file)

ann = {"anonymous-storage-FURL": furl,

"permutation-seed-base32": self._init_permutation_seed(ss),

}

current_seqnum, current_nonce = self._sequencer()

for ic in self.introducer_clients:

ic.publish("storage", ann, current_seqnum, current_nonce, self._node_key)

d.addCallback(_publish)

d.addErrback(log.err, facility="tahoe.init",

level=log.BAD, umid="aLGBKw")

File: allmydata/web/root.py

class Root(rend.Page):

(...)

def __init__(self, client, clock=None, now=None):

(...)

try:

s = client.getServiceNamed("storage")

except KeyError:

s = None

(...)

class StorageServer(service.MultiService, Referenceable):

(…)

def get_stats(self):

# remember: RIStatsProvider requires that our return dict

# contains numeric values.

stats = { 'storage_server.allocated': self.allocated_size(), }

stats['storage_server.reserved_space'] = self.reserved_space

for category,ld in self.get_latencies().items():

for name,v in ld.items():

stats['storage_server.latencies.%s.%s' % (category, name)] = v

try:

disk = fileutil.get_disk_stats(self.sharedir, self.reserved_space)

writeable = disk['avail'] > 0

# spacetime predictors should use disk_avail / (d(disk_used)/dt)

stats['storage_server.disk_total'] = disk['total']

stats['storage_server.disk_used'] = disk['used']

stats['storage_server.disk_free_for_root'] = disk['free_for_root']

stats['storage_server.disk_free_for_nonroot'] = disk['free_for_nonroot']

stats['storage_server.disk_avail'] = disk['avail']

except AttributeError:

writeable = True

except EnvironmentError:

log.msg("OS call to get disk statistics failed", level=log.UNUSUAL)

writeable = False

if self.readonly_storage:

stats['storage_server.disk_avail'] = 0

writeable = False

stats['storage_server.accepting_immutable_shares'] = int(writeable)

s = self.bucket_counter.get_state()

bucket_count = s.get("last-complete-bucket-count")

if bucket_count:

stats['storage_server.total_bucket_count'] = bucket_count

return stats

File: allmydata/immutable/checker.py

class Checker(log.PrefixingLogMixin):

"""I query all servers to see if M uniquely-numbered shares are

available.

(…)

def _get_buckets(self, s, storageindex):

"""Return a deferred that eventually fires with ({sharenum: bucket},

serverid, success). In case the server is disconnected or returns a

Failure then it fires with ({}, serverid, False) (A server

disconnecting or returning a Failure when we ask it for buckets is

the same, for our purposes, as a server that says it has none, except

that we want to track and report whether or not each server

responded.)"""

rref = s.get_rref()

lease_seed = s.get_lease_seed()

if self._add_lease:

renew_secret = self._get_renewal_secret(lease_seed)

cancel_secret = self._get_cancel_secret(lease_seed)

d2 = rref.callRemote("add_lease", storageindex,

renew_secret, cancel_secret)

d2.addErrback(self._add_lease_failed, s.get_name(), storageindex)

(...)

File: allmydata/mutable/servermap.py

class ServermapUpdater:

def __init__(self, filenode, storage_broker, monitor, servermap,

mode=MODE_READ, add_lease=False, update_range=None):

"""I update a servermap, locating a sufficient number of useful

shares and remembering where they are located.

"""

(…)

def _do_read(self, server, storage_index, shnums, readv):

ss = server.get_rref()

if self._add_lease:

# send an add-lease message in parallel. The results are handled

# separately. This is sent before the slot_readv() so that we can

# be sure the add_lease is retired by the time slot_readv comes

# back (this relies upon our knowledge that the server code for

# add_lease is synchronous).

renew_secret = self._node.get_renewal_secret(server)

cancel_secret = self._node.get_cancel_secret(server)

d2 = ss.callRemote("add_lease", storage_index,

renew_secret, cancel_secret)

# we ignore success

d2.addErrback(self._add_lease_failed, server, storage_index)

d = ss.callRemote("slot_readv", storage_index, shnums, readv)

return d

(...)

File: allmydata/scripts/tahoe_check.py

def check_location(options, where):

stdout = options.stdout

stderr = options.stderr

nodeurl = options['node-url']

if not nodeurl.endswith("/"):

nodeurl += "/"

try:

rootcap, path = get_alias(options.aliases, where, DEFAULT_ALIAS)

except UnknownAliasError, e:

e.display(stderr)

return 1

if path == '/':

path = ''

url = nodeurl + "uri/%s" % urllib.quote(rootcap)

if path:

url += "/" + escape_path(path)

# todo: should it end with a slash?

url += "?t=check&output=JSON"

if options["verify"]:

url += "&verify=true"

if options["repair"]:

url += "&repair=true"

if options["add-lease"]:

url += "&add-lease=true"

resp = do_http("POST", url)

if resp.status != 200:

print >>stderr, format_http_error("ERROR", resp)

return 1

jdata = resp.read()

if options.get("raw"):

stdout.write(jdata)

stdout.write("\n")

return 0

data = simplejson.loads(jdata)

File: allmydata/client.py

class Client(node.Node, pollmixin.PollMixin):

(…)

def init_client(self):

helper_furl = self.get_config("client", "helper.furl", None)

if helper_furl in ("None", ""):

helper_furl = None

DEP = self.encoding_params

DEP["k"] = int(self.get_config("client", "shares.needed", DEP["k"]))

DEP["n"] = int(self.get_config("client", "shares.total", DEP["n"]))

DEP["happy"] = int(self.get_config("client", "shares.happy", DEP["happy"]))

self.init_client_storage_broker()

self.history = History(self.stats_provider)

self.terminator = Terminator()

self.terminator.setServiceParent(self)

self.add_service(Uploader(helper_furl, self.stats_provider,

self.history))

File: allmydata/immutable/upload.py

class Uploader(service.MultiService, log.PrefixingLogMixin):

(...)

def __init__(self, helper_furl=None, stats_provider=None, history=None):

self._helper_furl = helper_furl

self.stats_provider = stats_provider

self._history = history

self._helper = None

self._all_uploads = weakref.WeakKeyDictionary() # for debugging

log.PrefixingLogMixin.__init__(self, facility="tahoe.immutable.upload")

service.MultiService.__init__(self)

def startService(self):

service.MultiService.startService(self)

if self._helper_furl:

self.parent.tub.connectTo(self._helper_furl,

self._got_helper)

def _got_helper(self, helper):

self.log("got helper connection, getting versions")

default = { "http://allmydata.org/tahoe/protocols/helper/v1" :

{ },

"application-version": "unknown: no get_version()",

}

d = add_version_to_remote_reference(helper, default)

d.addCallback(self._got_versioned_helper)

def _got_versioned_helper(self, helper):

needed = "http://allmydata.org/tahoe/protocols/helper/v1"

if needed not in helper.version:

raise InsufficientVersionError(needed, helper.version)

self._helper = helper

def _lost_helper(self):

self._helper = None

def get_helper_info(self):

# return a tuple of (helper_furl_or_None, connected_bool)

return (self._helper_furl, bool(self._helper))

File: allmydata/immutable/upload.py

class Uploader(service.MultiService, log.PrefixingLogMixin):

(...)

def upload(self, uploadable):

"""

Returns a Deferred that will fire with the UploadResults instance.

"""

assert self.parent

assert self.running

uploadable = IUploadable(uploadable)

d = uploadable.get_size()

def _got_size(size):

default_params = self.parent.get_encoding_parameters()

precondition(isinstance(default_params, dict), default_params)

precondition("max_segment_size" in default_params, default_params)

uploadable.set_default_encoding_parameters(default_params)

if self.stats_provider:

self.stats_provider.count('uploader.files_uploaded', 1)

self.stats_provider.count('uploader.bytes_uploaded', size)

if size <= self.URI_LIT_SIZE_THRESHOLD:

uploader = LiteralUploader()

return uploader.start(uploadable)

else:

eu = EncryptAnUploadable(uploadable, self._parentmsgid)

d2 = defer.succeed(None)

storage_broker = self.parent.get_storage_broker()

if self._helper:

uploader = AssistedUploader(self._helper, storage_broker)

d2.addCallback(lambda x: eu.get_storage_index())

d2.addCallback(lambda si: uploader.start(eu, si))

else:

storage_broker = self.parent.get_storage_broker()

secret_holder = self.parent._secret_holder

uploader = CHKUploader(storage_broker, secret_holder)

d2.addCallback(lambda x: uploader.start(eu))

self._all_uploads[uploader] = None

if self._history:

self._history.add_upload(uploader.get_upload_status())

def turn_verifycap_into_read_cap(uploadresults):

# Generate the uri from the verifycap plus the key.

d3 = uploadable.get_encryption_key()

def put_readcap_into_results(key):

v = uri.from_string(uploadresults.get_verifycapstr())

r = uri.CHKFileURI(key, v.uri_extension_hash, v.needed_shares, v.total_shares, v.size)

uploadresults.set_uri(r.to_string())

return uploadresults

d3.addCallback(put_readcap_into_results)

return d3

d2.addCallback(turn_verifycap_into_read_cap)

return d2

d.addCallback(_got_size)

def _done(res):

uploadable.close()

return res

d.addBoth(_done)

return d

File: allmydata/web/root.py

class Root(rend.Page):

def data_helper_furl_prefix(self, ctx, data):

try:

uploader = self.client.getServiceNamed("uploader")

except KeyError:

return None

furl, connected = uploader.get_helper_info()

if not furl:

return None

# trim off the secret swissnum

(prefix, _, swissnum) = furl.rpartition("/")

return "%s/[censored]" % (prefix,)

def data_helper_description(self, ctx, data):

if self.data_connected_to_helper(ctx, data) == "no":

return "Helper not connected"

return "Helper"

def data_connected_to_helper(self, ctx, data):

try:

uploader = self.client.getServiceNamed("uploader")

except KeyError:

return "no" # we don't even have an Uploader

furl, connected = uploader.get_helper_info()

if furl is None:

return "not-configured"

if connected:

return "yes"

return "no"

File: allmydata/web/welcome.xhtml

(…)

<div>

<h3>

<div><n:attr name="class">status-indicator connected-<n:invisible n:render="string" n:data="connected_to_helper" /></n:attr></div>

<div n:render="string" n:data="helper_description" />

</h3>

<div class="furl" n:render="string" n:data="helper_furl_prefix" />

</div>

(…)

File: allmydata/test/test_helper.py

class AssistedUpload(unittest.TestCase):

(...)

def setUpHelper(self, basedir, helper_class=Helper_fake_upload):

fileutil.make_dirs(basedir)

self.helper = h = helper_class(basedir,

self.s.storage_broker,

self.s.secret_holder,

None, None)

self.helper_furl = self.tub.registerReference(h)

def test_one(self):

self.basedir = "helper/AssistedUpload/test_one"

self.setUpHelper(self.basedir)

u = upload.Uploader(self.helper_furl)

u.setServiceParent(self.s)

d = wait_a_few_turns()

def _ready(res):

assert u._helper

return upload_data(u, DATA, convergence="some convergence string")

d.addCallback(_ready)

(…)

def test_previous_upload_failed(self):

(...)

f = open(encfile, "wb")

f.write(encryptor.process(DATA))

f.close()

u = upload.Uploader(self.helper_furl)

u.setServiceParent(self.s)

d = wait_a_few_turns()

def _ready(res):

assert u._helper

return upload_data(u, DATA, convergence="test convergence string")

d.addCallback(_ready)

(…)

def test_already_uploaded(self):

self.basedir = "helper/AssistedUpload/test_already_uploaded"

self.setUpHelper(self.basedir, helper_class=Helper_already_uploaded)

u = upload.Uploader(self.helper_furl)

u.setServiceParent(self.s)

d = wait_a_few_turns()

File: allmydata/client.py

class Client(node.Node, pollmixin.PollMixin):

(…)

def __init__(self, basedir="."):

node.Node.__init__(self, basedir)

self.started_timestamp = time.time()

self.logSource="Client"

self.encoding_params = self.DEFAULT_ENCODING_PARAMETERS.copy()

self.init_introducer_clients()

self.init_stats_provider()

self.init_secrets()

self.init_node_key()

self.init_storage()

(…)

def init_introducer_clients(self):

self.introducer_furls = []

self.warn_flag = False

# Try to load ""BASEDIR/introducers" cfg file

cfg = os.path.join(self.basedir, MULTI_INTRODUCERS_CFG)

if os.path.exists(cfg):

f = open(cfg, 'r')

for introducer_furl in f.read().split('\n'):

introducers_furl = introducer_furl.strip()

if introducers_furl.startswith('#') or not introducers_furl:

continue

self.introducer_furls.append(introducer_furl)

f.close()

furl_count = len(self.introducer_furls)

#print "@icfg: furls: %d" %furl_count

# read furl from tahoe.cfg

ifurl = self.get_config("client", "introducer.furl", None)

if ifurl and ifurl not in self.introducer_furls:

self.introducer_furls.append(ifurl)

f = open(cfg, 'a')

f.write(ifurl)

f.write('\n')

f.close()

if furl_count > 1:

self.warn_flag = True

self.log("introducers config file modified.")

print "Warning! introducers config file modified."

# create a pool of introducer_clients

self.introducer_clients = []

File: allmydata/storage_client.py

class StorageFarmBroker:

implements(IStorageBroker)

"""I live on the client, and know about storage servers. For each server

that is participating in a grid, I either maintain a connection to it or

remember enough information to establish a connection to it on demand.

I'm also responsible for subscribing to the IntroducerClient to find out

about new servers as they are announced by the Introducer.

"""

(...)

def use_introducer(self, introducer_client):

self.introducer_client = ic = introducer_client

ic.subscribe_to("storage", self._got_announcement)

def _got_announcement(self, key_s, ann):

if key_s is not None:

precondition(isinstance(key_s, str), key_s)

precondition(key_s.startswith("v0-"), key_s)

assert ann["service-name"] == "storage"

s = NativeStorageServer(key_s, ann)

serverid = s.get_serverid()

old = self.servers.get(serverid)

if old:

if old.get_announcement() == ann:

return # duplicate

# replacement

del self.servers[serverid]

old.stop_connecting()

# now we forget about them and start using the new one

self.servers[serverid] = s

s.start_connecting(self.tub, self._trigger_connections)

# the descriptor will manage their own Reconnector, and each time we

# need servers, we'll ask them if they're connected or not.

def _trigger_connections(self):

# when one connection is established, reset the timers on all others,

# to trigger a reconnection attempt in one second. This is intended

# to accelerate server connections when we've been offline for a

# while. The goal is to avoid hanging out for a long time with

# connections to only a subset of the servers, which would increase

# the chances that we'll put shares in weird places (and not update

(...)

File: allmydata/client.py

class Client(node.Node, pollmixin.PollMixin):

implements(IStatsProducer)

(…)

def init_storage(self):

# should we run a storage server (and publish it for others to use)?

if not self.get_config("storage", "enabled", True, boolean=True):

return

readonly = self.get_config("storage", "readonly", False, boolean=True)

storedir = os.path.join(self.basedir, self.STOREDIR)

(..)

ss = StorageServer(storedir, self.nodeid,

reserved_space=reserved,

discard_storage=discard,

readonly_storage=readonly,

stats_provider=self.stats_provider,

expiration_enabled=expire,

expiration_mode=mode,

expiration_override_lease_duration=o_l_d,

expiration_cutoff_date=cutoff_date,

expiration_sharetypes=expiration_sharetypes)

self.add_service(ss)

d = self.when_tub_ready()

# we can't do registerReference until the Tub is ready

def _publish(res):

furl_file = os.path.join(self.basedir, "private", "storage.furl").encode(get_filesystem_encoding())

furl = self.tub.registerReference(ss, furlFile=furl_file)

ann = {"anonymous-storage-FURL": furl,

"permutation-seed-base32": self._init_permutation_seed(ss),

}

current_seqnum, current_nonce = self._sequencer()

for ic in self.introducer_clients:

ic.publish("storage", ann, current_seqnum, current_nonce, self._node_key)

d.addCallback(_publish)

d.addErrback(log.err, facility="tahoe.init",

level=log.BAD, umid="aLGBKw")

File: allmydata/client.py

class Client(node.Node, pollmixin.PollMixin):

implements(IStatsProducer)

(…)

def init_helper(self):

d = self.when_tub_ready()

def _publish(self):

self.helper = Helper(os.path.join(self.basedir, "helper"),

self.storage_broker, self._secret_holder,

self.stats_provider, self.history)

# TODO: this is confusing. BASEDIR/private/helper.furl is created

# by the helper. BASEDIR/helper.furl is consumed by the client

# who wants to use the helper. I like having the filename be the

# same, since that makes 'cp' work smoothly, but the difference

# between config inputs and generated outputs is hard to see.

helper_furlfile = os.path.join(self.basedir,

"private", "helper.furl").encode(get_filesystem_encoding())

self.tub.registerReference(self.helper, furlFile=helper_furlfile)

d.addCallback(_publish)

d.addErrback(log.err, facility="tahoe.init",

level=log.BAD, umid="K0mW5w")

File: allmydata/introducer/server.py

class IntroducerNode(node.Node):

PORTNUMFILE = "introducer.port"

NODETYPE = "introducer"

GENERATED_FILES = ['introducer.furl']

def __init__(self, basedir="."):

node.Node.__init__(self, basedir)

self.read_config()

self.init_introducer()

webport = self.get_config("node", "web.port", None)

if webport:

self.init_web(webport) # strports string

def init_introducer(self):

introducerservice = IntroducerService(self.basedir)

self.add_service(introducerservice)

old_public_fn = os.path.join(self.basedir, "introducer.furl").encode(get_filesystem_encoding())

private_fn = os.path.join(self.basedir, "private", "introducer.furl").encode(get_filesystem_encoding())

(…)

d = self.when_tub_ready()

def _publish(res):

furl = self.tub.registerReference(introducerservice,

furlFile=private_fn)

self.log(" introducer is at %s" % furl, umid="qF2L9A")

self.introducer_url = furl # for tests

d.addCallback(_publish)

d.addErrback(log.err, facility="tahoe.init",

level=log.BAD, umid="UaNs9A")

(…)

class IntroducerService(service.MultiService, Referenceable):

implements(RIIntroducerPublisherAndSubscriberService_v2)

(…)

def publish(self, ann_t, canary, lp):

try:

self._publish(ann_t, canary, lp)

except:

log.err(format="Introducer.remote_publish failed on %(ann)s",

ann=ann_t,

level=log.UNUSUAL, parent=lp, umid="620rWA")

raise

(…)