From: Brian Warner <warner@allmydata.com>
Date: Fri, 10 Aug 2007 01:30:24 +0000 (-0700)
Subject: #96: add flag to enable pushing data to ourselves, defaulting to False
X-Git-Tag: allmydata-tahoe-0.5.0~83
X-Git-Url: https://git.rkrishnan.org/%5B/%5D%20/uri/-?a=commitdiff_plain;h=998802fd6d0977640469063a3cdaa26f3389a50f;p=tahoe-lafs%2Ftahoe-lafs.git
#96: add flag to enable pushing data to ourselves, defaulting to False
---
diff --git a/src/allmydata/client.py b/src/allmydata/client.py
index e54e195e..8f31a9bb 100644
--- a/src/allmydata/client.py
+++ b/src/allmydata/client.py
@@ -1,5 +1,6 @@
import os, sha, stat, time, re
+from base64 import b32decode
from foolscap import Referenceable, SturdyRef
from zope.interface import implements
from allmydata.interfaces import RIClient
@@ -29,6 +30,7 @@ class Client(node.Node, Referenceable):
MY_FURL_FILE = "myself.furl"
SUICIDE_PREVENTION_HOTLINE_FILE = "suicide_prevention_hotline"
SIZELIMIT_FILE = "sizelimit"
+ PUSH_TO_OURSELVES_FILE = "push_to_ourselves"
# we're pretty narrow-minded right now
OLDEST_SUPPORTED_VERSION = allmydata.__version__
@@ -38,6 +40,7 @@ class Client(node.Node, Referenceable):
self.my_furl = None
self.introducer_client = None
self.init_storage()
+ self.init_options()
self.add_service(Uploader())
self.add_service(Downloader())
self.add_service(VirtualDrive())
@@ -87,6 +90,12 @@ class Client(node.Node, Referenceable):
no_storage = os.path.exists(NOSTORAGE_FILE)
self.add_service(StorageServer(storedir, sizelimit, no_storage))
+ def init_options(self):
+ self.push_to_ourselves = None
+ filename = os.path.join(self.basedir, self.PUSH_TO_OURSELVES_FILE)
+ if os.path.exists(filename):
+ self.push_to_ourselves = True
+
def _check_hotline(self, hotline_file):
if os.path.exists(hotline_file):
mtime = os.stat(hotline_file)[stat.ST_MTIME]
@@ -141,18 +150,25 @@ class Client(node.Node, Referenceable):
return []
return self.introducer_client.get_all_peerids()
- def get_permuted_peers(self, key):
+ def get_permuted_peers(self, key, include_myself=True):
"""
@return: list of (permuted-peerid, peerid, connection,)
"""
results = []
+ myid = b32decode(self.tub.tubID.upper())
for peerid, connection in self.introducer_client.get_all_peers():
assert isinstance(peerid, str)
+ if not include_myself and peerid == myid:
+ self.log("get_permuted_peers: removing myself from the list")
+ continue
permuted = bytes_to_long(sha.new(key + peerid).digest())
results.append((permuted, peerid, connection))
results.sort()
return results
+ def get_push_to_ourselves(self):
+ return self.push_to_ourselves
+
def get_encoding_parameters(self):
if not self.introducer_client:
return None
diff --git a/src/allmydata/introducer.py b/src/allmydata/introducer.py
index 4d2f088f..2e889eb2 100644
--- a/src/allmydata/introducer.py
+++ b/src/allmydata/introducer.py
@@ -112,8 +112,8 @@ class IntroducerClient(service.Service, Referenceable):
self.log(" introducing ourselves: %s, %s" % (self, self.my_furl))
self._connected = True
d = introducer.callRemote("hello",
- node=self,
- furl=self.my_furl)
+ node=self,
+ furl=self.my_furl)
introducer.notifyOnDisconnect(self._disconnected)
def _disconnected(self):
diff --git a/src/allmydata/node.py b/src/allmydata/node.py
index 4c24f18d..8c99a89f 100644
--- a/src/allmydata/node.py
+++ b/src/allmydata/node.py
@@ -36,6 +36,8 @@ class Node(service.MultiService):
self.tub = Tub(certFile=certfile)
self.tub.setOption("logLocalFailures", True)
self.tub.setOption("logRemoteFailures", True)
+ # I think self.nodeid is kind of whacked. Shouldn't it equal the
+ # fingerprint portion of our furl?
self.nodeid = b32encode(self.tub.tubID).lower()
f = open(os.path.join(self.basedir, self.NODEIDFILE), "w")
f.write(b32encode(self.nodeid).lower() + "\n")
diff --git a/src/allmydata/test/check_memory.py b/src/allmydata/test/check_memory.py
index 71e0246a..604647c3 100644
--- a/src/allmydata/test/check_memory.py
+++ b/src/allmydata/test/check_memory.py
@@ -161,6 +161,10 @@ this file are ignored.
f = open(os.path.join(clientdir, "debug_no_storage"), "w")
f.write("no_storage\n")
f.close()
+ if self.mode == "upload-self":
+ f = open(os.path.join(clientdir, "push_to_ourselves"), "w")
+ f.write("push_to_ourselves\n")
+ f.close()
self.keepalive_file = os.path.join(clientdir,
"suicide_prevention_hotline")
# now start updating the mtime.
@@ -273,7 +277,7 @@ this file are ignored.
name = '%d' % size
print
print "uploading %s" % name
- if self.mode == "upload":
+ if self.mode in ("upload", "upload-self"):
files[name] = self.create_data(name, size)
d = control.callRemote("upload_from_file_to_uri", files[name])
def _done(uri):
diff --git a/src/allmydata/test/test_upload.py b/src/allmydata/test/test_upload.py
index 15148085..48ece345 100644
--- a/src/allmydata/test/test_upload.py
+++ b/src/allmydata/test/test_upload.py
@@ -129,9 +129,11 @@ class FakeBucketWriter:
class FakeClient:
def __init__(self, mode="good"):
self.mode = mode
- def get_permuted_peers(self, storage_index):
+ def get_permuted_peers(self, storage_index, include_myself):
return [ ("%20d"%fakeid, "%20d"%fakeid, FakePeer(self.mode),)
for fakeid in range(50) ]
+ def get_push_to_ourselves(self):
+ return None
def get_encoding_parameters(self):
return None
diff --git a/src/allmydata/upload.py b/src/allmydata/upload.py
index 65bcd2dc..055e82cc 100644
--- a/src/allmydata/upload.py
+++ b/src/allmydata/upload.py
@@ -88,7 +88,8 @@ class Tahoe3PeerSelector:
def get_shareholders(self, client,
storage_index, share_size, block_size,
- num_segments, total_shares, shares_of_happiness):
+ num_segments, total_shares, shares_of_happiness,
+ push_to_ourselves):
"""
@return: a set of PeerTracker instances that have agreed to hold some
shares for us
@@ -99,8 +100,9 @@ class Tahoe3PeerSelector:
# we are responsible for locating the shareholders. self._encoder is
# responsible for handling the data and sending out the shares.
- peers = client.get_permuted_peers(storage_index)
- assert peers
+ peers = client.get_permuted_peers(storage_index, push_to_ourselves)
+
+ assert peers, "peer selection left us with zero peers for our data"
# this needed_hashes computation should mirror
# Encoder.send_all_share_hash_trees. We use an IncompleteHashTree
@@ -398,10 +400,11 @@ class CHKUploader:
block_size = encoder.get_param("block_size")
num_segments = encoder.get_param("num_segments")
k,desired,n = encoder.get_param("share_counts")
+ push_to_ourselves = self._options.get("push_to_ourselves", False)
gs = peer_selector.get_shareholders
d = gs(self._client, storage_index, share_size, block_size,
- num_segments, n, desired)
+ num_segments, n, desired, push_to_ourselves)
return d
def set_shareholders(self, used_peers, encoder):
@@ -554,6 +557,10 @@ class Uploader(service.MultiService):
# this returns the URI
assert self.parent
assert self.running
+ push_to_ourselves = self.parent.get_push_to_ourselves()
+ if push_to_ourselves is not None:
+ options["push_to_ourselves"] = push_to_ourselves
+
uploadable = IUploadable(uploadable)
d = uploadable.get_size()
def _got_size(size):