Merge lp:~gholt/swift/lobjects3 into lp:~hudson-openstack/swift/trunk

[gholt]

Very large object support.

Very large object support.

Please read doc/source/overview_very_large_objects.rst to start.

[lp:~gholt/swift/lobjects3]

120. By gholt on 2010-11-02

Make poor Hudson happier

121. By gholt on 2010-11-02

Added missing SkipTest import

122. By gholt on 2010-11-02

Merged from trunk

123. By gholt on 2010-11-05

Merged from trunk

124. By gholt on 2010-11-08

Merge from trunk

[gholt (gholt)]

After a thorough discussion, this idea of transparent large object support is going to be dropped in favor of a combination client and cluster side solution.

review: Disapprove

Unmerged revisions

124. By gholt on 2010-11-08

Merge from trunk

123. By gholt on 2010-11-05

Merged from trunk

122. By gholt on 2010-11-02

Merged from trunk

121. By gholt on 2010-11-02

Added missing SkipTest import

120. By gholt on 2010-11-02

Make poor Hudson happier

119. By gholt on 2010-11-02

Documentation

118. By gholt on 2010-11-02

Tests for proxy server

117. By gholt on 2010-11-02

More object server tests

116. By gholt on 2010-11-02

More tests for the janitor

115. By gholt on 2010-11-02

Working on tests and bugfixes

=== modified file 'bin/swift-init'
--- bin/swift-init	2010-08-19 20:01:44 +0000
+++ bin/swift-init	2010-11-08 18:51:48 +0000
@@ -23,10 +23,11 @@
 import sys
 import time
 
-ALL_SERVERS = ['account-auditor', 'account-server', 'container-auditor',
+ALL_SERVERS = ['account-auditor', 'account-reaper', 'account-replicator',
+    'account-server', 'auth-server', 'container-auditor',
     'container-replicator', 'container-server', 'container-updater',
-    'object-auditor', 'object-server', 'object-replicator', 'object-updater',
-    'proxy-server', 'account-replicator', 'auth-server', 'account-reaper']
+    'object-auditor', 'object-janitor', 'object-replicator', 'object-server',
+    'object-updater', 'proxy-server']
 GRACEFUL_SHUTDOWN_SERVERS = ['account-server', 'container-server',
     'object-server', 'proxy-server', 'auth-server']
 MAX_DESCRIPTORS = 32768
@@ -41,6 +42,7 @@
     servers = [server]
 command = command.lower()
 
+
 def pid_files(server):
     if os.path.exists('/var/run/swift/%s.pid' % server):
         pid_files = ['/var/run/swift/%s.pid' % server]
@@ -50,6 +52,7 @@
         pid = int(open(pid_file).read().strip())
         yield pid_file, pid
 
+
 def do_start(server, once=False):
     server_type = '-'.join(server.split('-')[:-1])
 
@@ -77,7 +80,7 @@
                 os.makedirs(dir)
             except OSError, err:
                 if err.errno == errno.EACCES:
-                    sys.exit('Unable to create %s.  Running as non-root?' % dir)
+                    sys.exit('Unable to create %s. Running as non-root?' % dir)
         fp = open(pid_file, 'w')
         fp.write('%d\n' % pid)
         fp.close()
@@ -120,18 +123,21 @@
     elif os.path.exists('/etc/swift/%s-server/' % server_type):
         # found config directory, searching for config file(s)
         launch_args = []
-        for num, ini_file in enumerate(glob.glob('/etc/swift/%s-server/*.conf' % server_type)):
+        for num, ini_file in \
+             enumerate(glob.glob('/etc/swift/%s-server/*.conf' % server_type)):
             pid_file = '/var/run/swift/%s/%d.pid' % (server, num)
             # start a server for each ini_file found
             launch_args.append((ini_file, pid_file))
     else:
         # maybe there's a config file(s) out there, but I couldn't find it!
-        sys.exit('Unable to locate config file for %s. %s does not exist?' % (server, ini_file))
+        sys.exit('Unable to locate config file for %s. %s does not exist?' %
+                 (server, ini_file))
 
     # start all servers
     for ini_file, pid_file in launch_args:
         launch(ini_file, pid_file)
 
+
 def do_stop(server, graceful=False):
     if graceful and server in GRACEFUL_SHUTDOWN_SERVERS:
         sig = signal.SIGHUP

=== added file 'bin/swift-object-janitor'
--- bin/swift-object-janitor	1970-01-01 00:00:00 +0000
+++ bin/swift-object-janitor	2010-11-08 18:51:48 +0000
@@ -0,0 +1,28 @@
+#!/usr/bin/python
+# Copyright (c) 2010 OpenStack, LLC.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+# implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import sys
+
+from swift.obj.janitor import ObjectJanitor
+from swift.common import utils
+
+if __name__ == '__main__':
+    if len(sys.argv) < 2:
+        print "Usage: swift-object-janitor CONFIG_FILE [once]"
+        sys.exit(1)
+    once = len(sys.argv) > 2 and sys.argv[2] == 'once'
+    conf = utils.readconf(sys.argv[1], 'object-janitor')
+    ObjectJanitor(conf).run(once)

=== modified file 'doc/source/deployment_guide.rst'
--- doc/source/deployment_guide.rst	2010-10-21 18:59:43 +0000
+++ doc/source/deployment_guide.rst	2010-11-08 18:51:48 +0000
@@ -206,6 +206,25 @@
                                        object can be reclaimed
 ==================  =================  =======================================
 
+[object-janitor]
+
+===================  ==============  ==========================================
+Option               Default         Description
+-------------------  --------------  ------------------------------------------
+log_name             object-janitor  Label used when logging
+log_facility         LOG_LOCAL0      Syslog log facility
+log_level            INFO            Logging level
+interval             300             Minimum time for a pass to take
+concurrency          1               Number of updater workers to spawn
+node_timeout         10              Request timeout to external services
+conn_timeout         0.5             Connection timeout to external services
+slowdown             0.01            Time in seconds to wait between operations
+segment_reclaim_age  604800          Number of seconds before removing orphaned
+                                     object segments
+segments_per_pass    10              Maximum segments per object to remove per
+                                     pass
+===================  ==============  ==========================================
+
 [object-updater]
 
 ==================  ==============  ==========================================
@@ -438,6 +457,13 @@
 log_level                     INFO             Log level
 log_headers                   True             If True, log headers in each
                                                request
+max_object_size               107374182400     Maximum size of any object in
+                                               the cluster. Note: The
+                                               segment_size value will cause
+                                               objects larger than that to be
+                                               split into segments
+segment_size                  2147483647       Objects will be split into file
+                                               segments no larger than this
 recheck_account_existence     60               Cache timeout in seconds to
                                                send memcached for account
                                                existence

=== modified file 'doc/source/development_saio.rst'
--- doc/source/development_saio.rst	2010-11-02 14:17:25 +0000
+++ doc/source/development_saio.rst	2010-11-08 18:51:48 +0000
@@ -440,6 +440,8 @@
         [object-replicator]
         vm_test_mode = yes
 
+        [object-janitor]
+
         [object-updater]
 
         [object-auditor]
@@ -461,6 +463,8 @@
         [object-replicator]
         vm_test_mode = yes
 
+        [object-janitor]
+
         [object-updater]
 
         [object-auditor]
@@ -482,6 +486,8 @@
         [object-replicator]
         vm_test_mode = yes
 
+        [object-janitor]
+
         [object-updater]
 
         [object-auditor]
@@ -503,6 +509,8 @@
         [object-replicator]
         vm_test_mode = yes
 
+        [object-janitor]
+
         [object-updater]
 
         [object-auditor]
@@ -571,6 +579,7 @@
         # Replace devauth with whatever your super_admin key is (recorded in
         # /etc/swift/auth-server.conf).
         swift-auth-recreate-accounts -K devauth
+        swift-init object-janitor start
         swift-init object-updater start
         swift-init container-updater start
         swift-init object-replicator start

=== modified file 'doc/source/index.rst'
--- doc/source/index.rst	2010-11-04 19:25:23 +0000
+++ doc/source/index.rst	2010-11-08 18:51:48 +0000
@@ -26,6 +26,7 @@
     overview_replication
     overview_stats
     ratelimit
+    overview_very_large_objects
 
 Development:
 

=== modified file 'doc/source/object.rst'
--- doc/source/object.rst	2010-07-19 16:25:18 +0000
+++ doc/source/object.rst	2010-11-08 18:51:48 +0000
@@ -24,6 +24,16 @@
     :undoc-members:
     :show-inheritance:
 
+.. _object-janitor:
+
+Object Janitor
+==============
+
+.. automodule:: swift.obj.janitor
+    :members:
+    :undoc-members:
+    :show-inheritance:
+
 .. _object-updater:
 
 Object Updater
@@ -44,3 +54,12 @@
     :undoc-members:
     :show-inheritance:
 
+.. _object-diskfile:
+
+Disk File
+=========
+
+.. automodule:: swift.obj.diskfile
+    :members:
+    :undoc-members:
+    :show-inheritance:

=== added file 'doc/source/overview_very_large_objects.rst'
--- doc/source/overview_very_large_objects.rst	1970-01-01 00:00:00 +0000
+++ doc/source/overview_very_large_objects.rst	2010-11-08 18:51:48 +0000
@@ -0,0 +1,144 @@
+=========================
+Very Large Object Support
+=========================
+
+-----
+Intro
+-----
+
+Supporting very large objects in Swift presented quite a challenge. The main
+problem is storage balance; if you just have a few very large objects in the
+cluster some of the storage nodes will have significantly more data than the
+others. The basic answer is to break these objects up into segments and
+distribute the segments across the cluster evenly.
+
+The user could do this themselves, breaking up their very large objects into
+smaller objects and uploading those. But then they'd have to reassemble them
+themselves as well on download. What we do in Swift is essentially emulate this
+behavior for the user, transparently to the user. In this way, we can pick the
+optimal segment size for our cluster, something the user probably wouldn't
+know. This also allows the user to easily do ranged requests on the object
+without having to know how it's split up behind the scenes.
+
+-----------------------
+The Proxy is in Control
+-----------------------
+
+In Swift's implementation, the proxy server is in control of the object
+segmentation. As the user uploads a very large object, the proxy automaticaly
+closes the current segment when it reaches a configurable segment size and
+opens a new segment for more data. Once all the data is uploaded and stored,
+the proxy server then creates a manifest object indicating how the object was
+segmented so that it can be retreived later. The proxy does not spool any data
+to disk, it sends all data directly to the object servers.
+
+Because of these segment switchovers occurring in the proxy server, a very
+large object operation will have a higher chance of failure than a normal sized
+object. If at any point the proxy can't communicate with at least a majority of
+the nodes for each and every segment of the object, the whole object operation
+will fail. For instance, a regular 1m object in a 3 replica cluster just needs
+to make at least 2 object server requests to succeed. If the segment size is
+1G, a 10G object would need 2 object server requests to succeed 11 times during
+the operation (10 segments, 1 manifest).
+
+The upside of the proxy being in control is that the client does not need to
+know anything about the segmentation; it's done for them. The downside is that
+failure to upload means starting the whole thing over again. Later we plan to
+implement what we term 'user manifests' where the user can upload several
+objects but download them as if they were one. This would allow the user to
+just reupload a smaller part of the whole in the case of upload failures, but
+would require them to manage the segments themselves.
+
+-------------------------
+How They're Really Stored
+-------------------------
+
+If you have a working knowledge of how Swift stores and locates regular objects
+you'll know that the object name is hashed and that that hash determines the
+storage nodes and disk location for the object's data. Segmented objects work
+much the same way, but they have as many names as they do segments, plus one
+more for the manifest.
+
+The manifest is stored at the same location as the object would be if it were a
+normal object. The manifest is the last item stored once all the segments are
+in place, so that only one version of an object is available at any given time.
+This is important because you don't want to be able to retrieve a manifest and
+not be able to find the segments yet.
+
+The segments are given the same name as the object with the operation's
+timestamp and the segment number appended. The reason the timestamp is used is
+to keep the segments of multiple uploads of the same very large object from
+colliding. During an overwrite of a very large object, you don't want a
+download of that object to have some new data and some old. Since the segments
+for each version of the object are stored independently, a download before the
+new manifest file is in place will get the old data and a download afterwards
+will get the new data. Also, in the case where the new upload fails in transit,
+the old version will still be available.
+
+The storage nodes for each segment are determined the familiar way of hashing
+the segment name (object name + timestamp + segment number as described above)
+and then picking the storage nodes based on the hash. However, there is an
+exception for the first segment. This segment will always be stored on the same
+storage node and disk partition as a regular object of the same name would be,
+though the hash of the segment name determines the exact on disk location. This
+is because with chunked transfer encoded uploads we don't know the object size
+until the upload is complete. We won't realize we need to segment such an
+object until we've already uploaded the first segment's worth to the regular
+object storage nodes.
+
+To help keep directory structures smaller, object segments are stored in an
+'object_segments' directory alongside the usual 'objects' directory. Though all
+these files could be stored in the same directory structure, making this split
+should make replication, reclamation, and other scan-type operations less of a
+jar to the systems.
+
+.. note::
+
+    Because the first segment is always stored on the same storage nodes and in
+    the same partition as a regular file would, you can often see hashes in a
+    partition where they don't seem to belong. For instance, a first segment
+    name that hashes to 25d40100fd07c8c5fd1a4e31875593e1 might be found in
+    partition 568363 (hex 8ac2b) instead of the expected 154944 (hex 25d40)
+    because the segment's true object name hashed to
+    8ac2bf59556b61bb5cc521ccb51c200a. You should only see these anomalies in
+    the the object_segments directory, however.
+
+-----------
+Cleaning Up
+-----------
+
+Another challenge very large object support brought was in how to keep
+everything cleaned up and reduce the chance of completely orphaned segments
+that would waste a large amount of storage.
+
+One such scenario is the common DELETE operation. We don't want to make the
+client wait while we go out and delete all the segments and the manifest from
+all over the cluster. Instead, we write out a file that will start a background
+operation to the cleanup and we just remove the manifest file. An object
+overwrite is much like a delete in that just before placing the new manifest
+file we create the background job.
+
+Another scenario is failed uploads. If a user disconnects after uploading 100
+segments before finishing, we have to clean those up. Also, a proxy could crash
+due to hardware failure right in the middle of such an operation, so we
+wouldn't even be able to make a cleanup job at that point. So, we make a
+pending cleanup job just before starting any segmented upload. That pending
+cleanup job will wait up to a configurable amount of time (1 week by default)
+for the associated manifest to appear. If the manifest appears, the job is
+canceled and deleted. If the manifest never appears, the job goes about
+removing the orphaned segments from the system.
+
+Prior to very large object support, we had a background process called object
+async. Object async would send object metadata updates to the container servers
+in the event the container servers couldn't be contacted right away. These
+update jobs were uncommon and quick to complete, so durability wasn't too much
+of a concern. However, these segment cleanup jobs can hang around for a while
+(1 week in the above example of a proxy crash), so durable background jobs
+became important.
+
+With this, the object janitor was born. He accomplishes what object async did
+and the segment cleanup operations as well. The janitor jobs are stored just
+like regular object data is, except that they're stored in the object_janitor
+directory structure. Because they're stored in the same way, they can use the
+same object replicator code to keep durability with replicas on multiple
+storage nodes.

=== modified file 'etc/object-server.conf-sample'
--- etc/object-server.conf-sample	2010-10-19 15:02:36 +0000
+++ etc/object-server.conf-sample	2010-11-08 18:51:48 +0000
@@ -44,6 +44,21 @@
 # The replicator also performs reclamation
 # reclaim_age = 604800
 
+[object-janitor]
+# log_name = object-janitor
+# interval = 300
+# concurrency = 1
+# node_timeout = 10
+# conn_timeout = 0.5
+# slowdown will sleep that amount between janitor operations
+# slowdown = 0.01
+# Number of seconds before assuming a segmented put will never succeed and
+# therefore clean up any orphaned segments from the operation.
+# segment_reclaim_age = 604800
+# Number of segments to remove per pass when cleaning up after superceded or
+# orphaned segmented put operations.
+# segments_per_pass = 10
+
 [object-updater]
 # log_name = object-updater
 # interval = 300

=== modified file 'etc/proxy-server.conf-sample'
--- etc/proxy-server.conf-sample	2010-11-03 20:17:27 +0000
+++ etc/proxy-server.conf-sample	2010-11-08 18:51:48 +0000
@@ -17,6 +17,9 @@
 # log_facility = LOG_LOCAL0
 # log_level = INFO
 # log_headers = False
+# max_object_size = 107374182400
+# Files will be split into segments no larger than segment_size
+# segment_size = 2147483647
 # recheck_account_existence = 60
 # recheck_container_existence = 60
 # object_chunk_size = 8192

=== modified file 'setup.py'
--- setup.py	2010-11-03 19:50:35 +0000
+++ setup.py	2010-11-08 18:51:48 +0000
@@ -21,6 +21,7 @@
 
 from swift import __version__ as version
 
+
 class local_sdist(sdist):
     """Customized sdist hook - builds the ChangeLog file from VC first"""
 
@@ -57,29 +58,21 @@
         'Environment :: No Input/Output (Daemon)',
         ],
     install_requires=[],  # removed for better compat
-    scripts=[
-        'bin/st', 'bin/swift-account-auditor',
-        'bin/swift-account-audit', 'bin/swift-account-reaper',
-        'bin/swift-account-replicator', 'bin/swift-account-server',
-        'bin/swift-auth-add-user',
-        'bin/swift-auth-recreate-accounts', 'bin/swift-auth-server',
-        'bin/swift-auth-update-reseller-prefixes',
-        'bin/swift-container-auditor',
-        'bin/swift-container-replicator',
-        'bin/swift-container-server', 'bin/swift-container-updater',
-        'bin/swift-drive-audit', 'bin/swift-get-nodes',
-        'bin/swift-init', 'bin/swift-object-auditor',
-        'bin/swift-object-info',
-        'bin/swift-object-replicator',
-        'bin/swift-object-server',
-        'bin/swift-object-updater', 'bin/swift-proxy-server',
-        'bin/swift-ring-builder', 'bin/swift-stats-populate',
-        'bin/swift-stats-report',
-        'bin/swift-bench',
-        'bin/swift-log-uploader',
-        'bin/swift-log-stats-collector',
-        'bin/swift-account-stats-logger',
-        ],
+    scripts=['bin/st', 'bin/swift-account-audit', 'bin/swift-account-auditor',
+        'bin/swift-account-reaper', 'bin/swift-account-replicator',
+        'bin/swift-account-server', 'bin/swift-account-stats-logger',
+        'bin/swift-auth-add-user', 'bin/swift-auth-recreate-accounts',
+        'bin/swift-auth-server', 'bin/swift-auth-update-reseller-prefixes',
+        'bin/swift-bench', 'bin/swift-container-auditor',
+        'bin/swift-container-replicator', 'bin/swift-container-server',
+        'bin/swift-container-updater', 'bin/swift-drive-audit',
+        'bin/swift-get-nodes', 'bin/swift-init',
+        'bin/swift-log-stats-collector', 'bin/swift-log-uploader',
+        'bin/swift-object-auditor', 'bin/swift-object-info',
+        'bin/swift-object-janitor', 'bin/swift-object-replicator',
+        'bin/swift-object-server', 'bin/swift-object-updater',
+        'bin/swift-proxy-server', 'bin/swift-ring-builder',
+        'bin/swift-stats-populate', 'bin/swift-stats-report'],
     entry_points={
         'paste.app_factory': [
             'proxy=swift.proxy.server:app_factory',

=== modified file 'swift/common/constraints.py'
--- swift/common/constraints.py	2010-10-26 15:13:14 +0000
+++ swift/common/constraints.py	2010-11-08 18:51:48 +0000
@@ -19,8 +19,6 @@
     HTTPRequestEntityTooLarge
 
 
-#: Max file size allowed for objects
-MAX_FILE_SIZE = 5 * 1024 * 1024 * 1024 + 2
 #: Max length of the name of a key for metadata
 MAX_META_NAME_LENGTH = 128
 #: Max length of the value of a key for metadata
@@ -29,14 +27,18 @@
 MAX_META_COUNT = 90
 #: Max overall size of metadata
 MAX_META_OVERALL_SIZE = 4096
+#: Max account name length
+MAX_ACCOUNT_NAME_LENGTH = 256
+#: Max container name length
+MAX_CONTAINER_NAME_LENGTH = 256
 #: Max object name length
 MAX_OBJECT_NAME_LENGTH = 1024
 #: Max object list length of a get request for a container
 CONTAINER_LISTING_LIMIT = 10000
 #: Max container list length of a get request for an account
 ACCOUNT_LISTING_LIMIT = 10000
-MAX_ACCOUNT_NAME_LENGTH = 256
-MAX_CONTAINER_NAME_LENGTH = 256
+#: Default pickle protocol number used for Swift pickles
+PICKLE_PROTOCOL = 2
 
 
 def check_metadata(req, target_type):
@@ -82,19 +84,22 @@
     return None
 
 
-def check_object_creation(req, object_name):
+def check_object_creation(req, object_name, max_object_size=0):
     """
     Check to ensure that everything is alright about an object to be created.
 
     :param req: HTTP request object
     :param object_name: name of object to be created
+    :param max_object_size: the maximum object size to check against; 0 if no
+                            object size checking should be done
     :raises HTTPRequestEntityTooLarge: the object is too large
     :raises HTTPLengthRequered: missing content-length header and not
                                 a chunked request
     :raises HTTPBadRequest: missing or bad content-type header, or
                             bad metadata
     """
-    if req.content_length and req.content_length > MAX_FILE_SIZE:
+    if max_object_size > 0 and req.content_length and \
+            req.content_length > max_object_size:
         return HTTPRequestEntityTooLarge(body='Your request is too large.',
                 request=req, content_type='text/plain')
     if req.content_length is None and \

=== modified file 'swift/common/db.py'
--- swift/common/db.py	2010-08-16 22:30:27 +0000
+++ swift/common/db.py	2010-11-08 18:51:48 +0000
@@ -32,6 +32,7 @@
 import simplejson as json
 import sqlite3
 
+from swift.common.constraints import PICKLE_PROTOCOL
 from swift.common.utils import normalize_timestamp, renamer, \
         mkdirs, lock_parent_directory, fallocate
 from swift.common.exceptions import LockTimeout
@@ -39,8 +40,6 @@
 
 #: Timeout for trying to connect to a DB
 BROKER_TIMEOUT = 25
-#: Pickle protocol to use
-PICKLE_PROTOCOL = 2
 #: Max number of pending entries
 PENDING_CAP = 131072
 

=== modified file 'swift/obj/auditor.py'
--- swift/obj/auditor.py	2010-10-21 18:32:10 +0000
+++ swift/obj/auditor.py	2010-11-08 18:51:48 +0000
@@ -18,8 +18,8 @@
 from hashlib import md5
 from random import random
 
-from swift.obj import server as object_server
-from swift.obj.replicator import invalidate_hash
+from swift.obj.diskfile import DATADIR, DiskFile, invalidate_hash, \
+                               read_metadata
 from swift.common.utils import get_logger, renamer, audit_location_generator
 from swift.common.exceptions import AuditException
 from swift.common.daemon import Daemon
@@ -45,10 +45,8 @@
         time.sleep(random() * self.interval)
         while True:
             begin = time.time()
-            all_locs = audit_location_generator(self.devices,
-                                                object_server.DATADIR,
-                                                mount_check=self.mount_check,
-                                                logger=self.logger)
+            all_locs = audit_location_generator(self.devices, DATADIR,
+                        mount_check=self.mount_check, logger=self.logger)
             for path, device, partition in all_locs:
                 self.object_audit(path, device, partition)
                 if time.time() - reported >= 3600:  # once an hour
@@ -68,10 +66,8 @@
         """Run the object audit once."""
         self.logger.info('Begin object audit "once" mode')
         begin = reported = time.time()
-        all_locs = audit_location_generator(self.devices,
-                                            object_server.DATADIR,
-                                            mount_check=self.mount_check,
-                                            logger=self.logger)
+        all_locs = audit_location_generator(self.devices, DATADIR,
+                    mount_check=self.mount_check, logger=self.logger)
         for path, device, partition in all_locs:
             self.object_audit(path, device, partition)
             if time.time() - reported >= 3600:  # once an hour
@@ -99,14 +95,12 @@
             if not path.endswith('.data'):
                 return
             try:
-                name = object_server.read_metadata(path)['name']
+                name = read_metadata(path)['name']
             except Exception, exc:
                 raise AuditException('Error when reading metadata: %s' % exc)
             _, account, container, obj = name.split('/', 3)
-            df = object_server.DiskFile(self.devices, device,
-                                        partition, account,
-                                        container, obj,
-                                        keep_data_fp=True)
+            df = DiskFile(self.devices, device, partition, account, container,
+                          obj, keep_data_fp=True)
             if df.data_file is None:
                 # file is deleted, we found the tombstone
                 return

=== added file 'swift/obj/diskfile.py'
--- swift/obj/diskfile.py	1970-01-01 00:00:00 +0000
+++ swift/obj/diskfile.py	2010-11-08 18:51:48 +0000
@@ -0,0 +1,507 @@
+# Copyright (c) 2010 OpenStack, LLC.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+# implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from __future__ import with_statement
+import cPickle as pickle
+import errno
+import hashlib
+import os
+from contextlib import contextmanager
+from os.path import basename, dirname, isdir, join, splitext
+from tempfile import mkstemp
+from time import time
+
+from eventlet import tpool, sleep
+from xattr import getxattr, setxattr
+
+from swift.common.constraints import PICKLE_PROTOCOL
+from swift.common.utils import drop_buffer_cache, hash_path, lock_path, \
+    mkdirs, normalize_timestamp, renamer, split_path, storage_directory
+
+
+DATADIR = 'objects'
+SEGMENTSDIR = 'object_segments'
+JANITORDIR = 'object_janitor'
+HASH_FILE = 'hashes.pkl'
+METADATA_KEY = 'user.swift.metadata'
+ONE_WEEK = 604800
+
+
+def read_metadata(fd):
+    """
+    Helper function to read the pickled metadata from an object file.
+
+    :param fd: file descriptor to load the metadata from
+
+    :returns: dictionary of metadata
+    """
+    metadata = ''
+    key = 0
+    try:
+        while True:
+            metadata += getxattr(fd, '%s%s' % (METADATA_KEY, (key or '')))
+            key += 1
+    except IOError:
+        pass
+    return pickle.loads(metadata)
+
+
+def hash_suffix(path, reclaim_age):
+    """
+    Performs reclamation and returns an md5 of all (remaining) files.
+
+    :param reclaim_age: age in seconds at which to remove tombstones
+    """
+    md5 = hashlib.md5()
+    for hsh in sorted(os.listdir(path)):
+        hsh_path = join(path, hsh)
+        files = os.listdir(hsh_path)
+        if len(files) == 1:
+            if files[0].endswith('.ts'):
+                # remove tombstones older than reclaim_age
+                ts = files[0].rsplit('.', 1)[0]
+                if (time() - float(ts)) > reclaim_age:
+                    os.unlink(join(hsh_path, files[0]))
+                    files.remove(files[0])
+        elif files:
+            files.sort(reverse=True)
+            meta = data = tomb = None
+            for filename in files:
+                if not meta and filename.endswith('.meta'):
+                    meta = filename
+                if not data and filename.endswith('.data'):
+                    data = filename
+                if not tomb and filename.endswith('.ts'):
+                    tomb = filename
+                if (filename < tomb or       # any file older than tomb
+                    filename < data or       # any file older than data
+                    (filename.endswith('.meta') and
+                     filename < meta)):      # old meta
+                    if filename.endswith('.data'):
+                        fp = open(join(hsh_path, filename), 'rb')
+                        metadata = read_metadata(fp)
+                        if metadata.get('X-Object-Type') == 'manifest':
+                            manifest = pickle.load(fp)
+                            partition = dirname(path)
+                            device = dirname(dirname(partition))
+                            devices = dirname(device)
+                            partition = basename(partition)
+                            device = basename(device)
+                            account, container, obj = \
+                                split_path(metadata['name'], 3,
+                                           rest_with_last=True)
+                            df = DiskFile(devices, device, partition,
+                                'Segment-Cleanup', manifest['x-timestamp'],
+                                '%s/%s/%s' % (account, container, obj),
+                                datadir=JANITORDIR)
+                            df.store_janitor_segment_cleanup(account,
+                                container, obj,
+                                segment_count=(manifest['content-length'] /
+                                               manifest['x-segment-size'] + 1),
+                                segment_last_deleted=None)
+                        fp.close()
+                    os.unlink(join(hsh_path, filename))
+                    files.remove(filename)
+        if not files:
+            os.rmdir(hsh_path)
+        for filename in files:
+            md5.update(filename)
+    try:
+        os.rmdir(path)
+    except OSError:
+        pass
+    return md5.hexdigest()
+
+
+def recalculate_hashes(partition_dir, suffixes, reclaim_age=ONE_WEEK):
+    """
+    Recalculates hashes for the given suffixes in the partition and updates
+    them in the partition's hashes file.
+
+    :param partition_dir: directory of the partition in which to recalculate
+    :param suffixes: list of suffixes to recalculate
+    :param reclaim_age: age in seconds at which tombstones should be removed
+    """
+
+    def tpool_listdir(partition_dir):
+        return dict(((suff, None) for suff in os.listdir(partition_dir)
+                     if len(suff) == 3 and isdir(join(partition_dir, suff))))
+    hashes_file = join(partition_dir, HASH_FILE)
+    with lock_path(partition_dir):
+        try:
+            with open(hashes_file, 'rb') as fp:
+                hashes = pickle.load(fp)
+        except Exception:
+            hashes = tpool.execute(tpool_listdir, partition_dir)
+        for suffix in suffixes:
+            suffix_dir = join(partition_dir, suffix)
+            if os.path.exists(suffix_dir):
+                hashes[suffix] = hash_suffix(suffix_dir, reclaim_age)
+            elif suffix in hashes:
+                del hashes[suffix]
+        with open(hashes_file + '.tmp', 'wb') as fp:
+            pickle.dump(hashes, fp, PICKLE_PROTOCOL)
+        renamer(hashes_file + '.tmp', hashes_file)
+
+
+def invalidate_hash(suffix_dir):
+    """
+    Invalidates the hash for a suffix_dir in the partition's hashes file.
+
+    :param suffix_dir: absolute path to suffix dir whose hash needs
+                       invalidating
+    """
+
+    suffix = os.path.basename(suffix_dir)
+    partition_dir = os.path.dirname(suffix_dir)
+    hashes_file = join(partition_dir, HASH_FILE)
+    with lock_path(partition_dir):
+        try:
+            with open(hashes_file, 'rb') as fp:
+                hashes = pickle.load(fp)
+            if suffix in hashes and not hashes[suffix]:
+                return
+        except Exception:
+            return
+        hashes[suffix] = None
+        with open(hashes_file + '.tmp', 'wb') as fp:
+            pickle.dump(hashes, fp, PICKLE_PROTOCOL)
+        renamer(hashes_file + '.tmp', hashes_file)
+
+
+def get_hashes(partition_dir, do_listdir=True, reclaim_age=ONE_WEEK):
+    """
+    Get a list of hashes for the suffix dir.  do_listdir causes it to mistrust
+    the hash cache for suffix existence at the (unexpectedly high) cost of a
+    listdir.  reclaim_age is just passed on to hash_suffix.
+
+    :param partition_dir: absolute path of partition to get hashes for
+    :param do_listdir: force existence check for all hashes in the partition
+    :param reclaim_age: age at which to remove tombstones
+
+    :returns: tuple of (number of suffix dirs hashed, dictionary of hashes)
+    """
+
+    def tpool_listdir(hashes, partition_dir):
+        return dict(((suff, hashes.get(suff, None))
+                     for suff in os.listdir(partition_dir)
+                     if len(suff) == 3 and isdir(join(partition_dir, suff))))
+    hashed = 0
+    hashes_file = join(partition_dir, HASH_FILE)
+    with lock_path(partition_dir):
+        modified = False
+        hashes = {}
+        try:
+            with open(hashes_file, 'rb') as fp:
+                hashes = pickle.load(fp)
+        except Exception:
+            do_listdir = True
+        if do_listdir:
+            hashes = tpool.execute(tpool_listdir, hashes, partition_dir)
+            modified = True
+        for suffix, hash_ in hashes.items():
+            if not hash_:
+                suffix_dir = join(partition_dir, suffix)
+                if os.path.exists(suffix_dir):
+                    try:
+                        hashes[suffix] = hash_suffix(suffix_dir, reclaim_age)
+                        hashed += 1
+                    except OSError:
+                        logging.exception('Error hashing suffix')
+                        hashes[suffix] = None
+                else:
+                    del hashes[suffix]
+                modified = True
+                sleep()
+        if modified:
+            with open(hashes_file + '.tmp', 'wb') as fp:
+                pickle.dump(hashes, fp, PICKLE_PROTOCOL)
+            renamer(hashes_file + '.tmp', hashes_file)
+        return hashed, hashes
+
+
+class DiskFile(object):
+    """
+    Manage files on disk for a single object.
+
+    :param path: path to devices on the node
+    :param device: device name
+    :param partition: partition on the device the object lives in
+    :param account: account name for the object
+    :param container: container name for the object
+    :param obj: object name for the object
+    :param keep_data_fp: if True, don't close the fp, otherwise close it
+    :param disk_chunk_size: size of chunks on file reads
+    :param datadir: Sets which directory the root of the data structure is
+                    named (default: DATADIR)
+    :param segment: If set to not None, indicates which segment of an object
+                    this file represents
+    :param segment_timestamp: X-Timestamp of the object's segments (set on the
+                              PUT, not changed on POSTs), required if segment
+                              is set to not None
+    """
+
+    def __init__(self, path, device, partition, account, container, obj,
+                 keep_data_fp=False, disk_chunk_size=65536, datadir=DATADIR,
+                 segment=None, segment_timestamp=None):
+        self.account = account
+        self.container = container
+        self.obj = obj
+        self.fp = None
+        self.disk_chunk_size = disk_chunk_size
+        self.name = '/' + '/'.join((account, container, obj))
+        if segment is not None:
+            segment_name = '%s/%s/%s' % (obj, segment_timestamp, segment)
+            segment_hash = hash_path(account, container, segment_name)
+            self.datadir = os.path.join(path, device,
+                storage_directory(SEGMENTSDIR, partition, segment_hash))
+            name_hash = hash_path(account, container, obj)
+            self.no_longer_segment_datadir = os.path.join(path, device,
+                storage_directory(datadir, partition, name_hash))
+        else:
+            name_hash = hash_path(account, container, obj)
+            self.datadir = self.no_longer_segment_datadir = os.path.join(path,
+                device, storage_directory(datadir, partition, name_hash))
+        self.tmpdir = os.path.join(path, device, 'tmp')
+        self.metadata = {}
+        self.meta_file = None
+        self.data_file = None
+        if not os.path.exists(self.datadir):
+            return
+        files = sorted(os.listdir(self.datadir), reverse=True)
+        for file in files:
+            if file.endswith('.ts'):
+                self.data_file = self.meta_file = None
+                self.metadata = {'deleted': True}
+                return
+            if file.endswith('.meta') and not self.meta_file:
+                self.meta_file = os.path.join(self.datadir, file)
+            if file.endswith('.data') and not self.data_file:
+                self.data_file = os.path.join(self.datadir, file)
+                break
+        if not self.data_file:
+            return
+        self.fp = open(self.data_file, 'rb')
+        self.metadata = read_metadata(self.fp)
+        if not keep_data_fp:
+            self.close()
+        if self.meta_file:
+            with open(self.meta_file) as mfp:
+                for key in self.metadata.keys():
+                    if key.lower() not in ('content-type', 'content-encoding',
+                                'deleted', 'content-length', 'etag'):
+                        del self.metadata[key]
+                self.metadata.update(read_metadata(mfp))
+
+    def __iter__(self):
+        """Returns an iterator over the data file."""
+        try:
+            dropped_cache = 0
+            read = 0
+            while True:
+                chunk = self.fp.read(self.disk_chunk_size)
+                if chunk:
+                    read += len(chunk)
+                    if read - dropped_cache > (1024 * 1024):
+                        drop_buffer_cache(self.fp.fileno(), dropped_cache,
+                            read - dropped_cache)
+                        dropped_cache = read
+                    yield chunk
+                else:
+                    drop_buffer_cache(self.fp.fileno(), dropped_cache,
+                        read - dropped_cache)
+                    break
+        finally:
+            self.close()
+
+    def app_iter_range(self, start, stop):
+        """Returns an iterator over the data file for range (start, stop)"""
+        if start:
+            self.fp.seek(start)
+        if stop is not None:
+            length = stop - start
+        else:
+            length = None
+        for chunk in self:
+            if length is not None:
+                length -= len(chunk)
+                if length < 0:
+                    # Chop off the extra:
+                    yield chunk[:length]
+                    break
+            yield chunk
+
+    def close(self):
+        """Close the file."""
+        if self.fp:
+            self.fp.close()
+            self.fp = None
+
+    def is_deleted(self):
+        """
+        Check if the file is deleted.
+
+        :returns: True if the file doesn't exist or has been flagged as
+                  deleted.
+        """
+        return not self.data_file or 'deleted' in self.metadata
+
+    @contextmanager
+    def mkstemp(self):
+        """Context manager to make a temporary file."""
+        if not os.path.exists(self.tmpdir):
+            mkdirs(self.tmpdir)
+        fd, tmppath = mkstemp(dir=self.tmpdir)
+        try:
+            yield fd, tmppath
+        finally:
+            try:
+                os.close(fd)
+            except OSError:
+                pass
+            try:
+                os.unlink(tmppath)
+            except OSError:
+                pass
+
+    def put(self, fd, tmppath, metadata, extension='.data',
+            no_longer_segment=False):
+        """
+        Finalize writing the file on disk, and renames it from the temp file to
+        the real location.  This should be called after the data has been
+        written to the temp file.
+
+        :params fd: file descriptor of the temp file
+        :param tmppath: path to the temporary file being used
+        :param metadata: dictionary of metada to be written
+        :param extension: extension to be used when making the file
+        :param no_longer_segment: Set to True if this was originally an object
+            segment but no longer is (case with chunked transfer encoding when
+            the object ends up less than the segment size)
+        """
+        metadata['name'] = self.name
+        timestamp = normalize_timestamp(metadata['X-Timestamp'])
+        metastr = pickle.dumps(metadata, PICKLE_PROTOCOL)
+        key = 0
+        while metastr:
+            setxattr(fd, '%s%s' % (METADATA_KEY, key or ''), metastr[:254])
+            metastr = metastr[254:]
+            key += 1
+        if 'Content-Length' in metadata:
+            drop_buffer_cache(fd, 0, int(metadata['Content-Length']))
+        os.fsync(fd)
+        if no_longer_segment:
+            self.datadir = self.no_longer_segment_datadir
+        invalidate_hash(os.path.dirname(self.datadir))
+        renamer(tmppath, os.path.join(self.datadir, timestamp + extension))
+        self.metadata = metadata
+
+    def unlinkold(self, timestamp):
+        """
+        Remove any older versions of the object file.  Any file that has an
+        older timestamp than timestamp will be deleted.
+
+        :param timestamp: timestamp to compare with each file
+        """
+        timestamp = normalize_timestamp(timestamp)
+        for fname in os.listdir(self.datadir):
+            if fname < timestamp:
+                try:
+                    os.unlink(os.path.join(self.datadir, fname))
+                except OSError, err:    # pragma: no cover
+                    if err.errno != errno.ENOENT:
+                        raise
+
+    def tombstone(self, timestamp):
+        """
+        Creates a tombstone for the DiskFile, indicating any versions older
+        than `timestamp` should be removed.
+
+        :param timestamp: normalized timestamp of the tombstone
+        """
+        with self.mkstemp() as (fd, tmppath):
+            self.put(fd, tmppath, {'X-Timestamp': timestamp, 'deleted': True},
+                     extension='.ts')
+        self.unlinkold(timestamp)
+
+    def store_janitor_container_update(self, op, account, container, obj,
+                                       headers, successes):
+        """
+        Creates a .data file whose contents contain a operation for the
+        object-janitor to send object metadata to the container servers.
+
+        :param op: The operation to send to the container server (usually PUT
+                   or DELETE).
+        :param account: The account name for the object.
+        :param container: The container name for the object.
+        :param obj: The object name for the object.
+        :param headers: The headers to include in the requests to the container
+                        servers. Should at least contain X-Timestamp indicating
+                        the version of the object metadata.
+        :param successes: An array of container node ids that have already
+                          received the object metadata update.
+        """
+        timestamp = normalize_timestamp(time())
+        with self.mkstemp() as (fd, tmppath):
+            os.write(fd, pickle.dumps({'op': op, 'account': account,
+                'container': container, 'obj': obj, 'headers': headers,
+                'successes': successes}, PICKLE_PROTOCOL))
+            self.put(fd, tmppath,
+                {'X-Op': 'Container-Update', 'X-Timestamp': timestamp})
+        self.unlinkold(timestamp)
+
+    def store_janitor_segment_cleanup(self, account, container, obj,
+                                      segment_count, segment_last_deleted):
+        """
+        Creates a .data file whose contents contain a operation for the
+        object-janitor to send clean up object segments for an object.
+
+        Note that the DiskFile created for this operation is a bit different
+        than most DiskFiles. The DiskFile account name should be
+        'Segment-Cleanup', the container name should be the segments'
+        timestamp, and the object name should be the full account/container/obj
+        path (no leading /). However, the account, container, obj given in this
+        `store_janitor_segment_cleanup` call should be the usual as related to
+        the actual object. This complexity is so that each set of segments are
+        treated independently.
+
+        For example::
+
+            df = DiskFile(devices, device, partition,
+                    'Segment-Cleanup', manifest['x-timestamp'],
+                    '%s/%s/%s' % (account, container, obj), datadir=JANITORDIR)
+            df.store_janitor_segment_cleanup(account, container, obj, None,
+                                             None)
+
+        :param account: The account name for the object.
+        :param container: The container name for the object.
+        :param obj: The object name for the object.
+        :param segment_count: The number of segments the object has or None if
+                              the number is not known.
+        :param segment_last_deleted: The segment that was last deleted so the
+                                     next pass of this operation can continue
+                                     where it left off. Set to None if no
+                                     segments have been deleted yet.
+        """
+        timestamp = normalize_timestamp(time())
+        with self.mkstemp() as (fd, tmppath):
+            os.write(fd, pickle.dumps({'account': account,
+                'container': container, 'obj': obj,
+                'segment_count': segment_count,
+                'segment_last_deleted': segment_last_deleted},
+                PICKLE_PROTOCOL))
+            self.put(fd, tmppath,
+                     {'X-Op': 'Segment-Cleanup', 'X-Timestamp': timestamp})
+        self.unlinkold(timestamp)

=== added file 'swift/obj/janitor.py'
--- swift/obj/janitor.py	1970-01-01 00:00:00 +0000
+++ swift/obj/janitor.py	2010-11-08 18:51:48 +0000
@@ -0,0 +1,516 @@
+# Copyright (c) 2010 OpenStack, LLC.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+# implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from __future__ import with_statement
+import cPickle as pickle
+import os
+import signal
+import sys
+from random import random
+from time import sleep, time
+
+from eventlet import patcher, Timeout
+
+from swift.common.bufferedhttp import http_connect
+from swift.common.exceptions import ConnectionTimeout
+from swift.common.ring import Ring
+from swift.common.utils import get_logger, normalize_timestamp, whataremyips
+from swift.common.daemon import Daemon
+from swift.obj.diskfile import DiskFile, JANITORDIR, read_metadata
+
+
+class ObjectJanitor(Daemon):
+    """
+    Run background operations for the object server, such as updating container
+    servers with new object metadata and cleaning up discarded segmented
+    objects.
+    """
+
+    def __init__(self, conf):
+        self.conf = conf
+        self.logger = get_logger(conf, 'object-janitor')
+        self.devices = conf.get('devices', '/srv/node')
+        self.port = int(conf.get('bind_port', 6000))
+        self.my_node_ids = []
+        self.mount_check = conf.get('mount_check', 'true').lower() in \
+                              ('true', 't', '1', 'on', 'yes', 'y')
+        swift_dir = conf.get('swift_dir', '/etc/swift')
+        self.interval = int(conf.get('interval', 300))
+        self.object_ring_path = os.path.join(swift_dir, 'object.ring.gz')
+        self.object_ring = None
+        self.container_ring_path = os.path.join(swift_dir, 'container.ring.gz')
+        self.container_ring = None
+        self.concurrency = int(conf.get('concurrency', 1))
+        self.slowdown = float(conf.get('slowdown', 0.01))
+        self.node_timeout = int(conf.get('node_timeout', 10))
+        self.conn_timeout = float(conf.get('conn_timeout', 0.5))
+        self.segment_reclaim_age = int(conf.get('segment_reclaim_age', 604800))
+        self.segments_per_pass = int(conf.get('segments_per_pass', 10))
+        self.container_update_successes = 0
+        self.container_update_failures = 0
+        self.segment_cleanup_completions = 0
+        self.segment_cleanup_segments = 0
+        self.segment_cleanup_failures = 0
+
+    def get_object_ring(self):
+        """Get the object ring.  Load it, if it hasn't been yet."""
+        if not self.object_ring:
+            self.logger.debug(
+                'Loading object ring from %s' % self.object_ring_path)
+            self.object_ring = Ring(self.object_ring_path)
+        return self.object_ring
+
+    def get_container_ring(self):
+        """Get the container ring.  Load it, if it hasn't been yet."""
+        if not self.container_ring:
+            self.logger.debug(
+                'Loading container ring from %s' % self.container_ring_path)
+            self.container_ring = Ring(self.container_ring_path)
+        return self.container_ring
+
+    def run_forever(self):
+        """Run the janitor continuously."""
+        sleep(random() * self.interval)
+        while True:
+            begin = time()
+            self.full_sweep()
+            elapsed = time() - begin
+            if elapsed < self.interval:
+                sleep(self.interval - elapsed)
+
+    def run_once(self):
+        """Run the janitor once."""
+        self.full_sweep(fork=False)
+
+    def full_sweep(self, fork=True):
+        """
+        Run a full sweep of the server for object janitor operations.
+
+        :param fork: If True, subprocesses will be forked for each device up to
+                     self.concurrency at any given time.
+        """
+        self.logger.info('Begin object janitor sweep')
+        begin = time()
+        my_ips = whataremyips()
+        self.my_node_ids = [n['id'] for n in self.get_object_ring().devs
+            if n and n['ip'] in my_ips and n['port'] == self.port]
+        pids = []
+        for device in os.listdir(self.devices):
+            if self.mount_check and not \
+                    os.path.ismount(os.path.join(self.devices, device)):
+                self.logger.warn('Skipping %s as it is not mounted' % device)
+                continue
+            if fork:
+                while len(pids) >= self.concurrency:
+                    pids.remove(os.wait()[0])
+            # read from rings to ensure they're fresh
+            self.get_object_ring().get_nodes('')
+            self.get_container_ring().get_nodes('')
+            if fork:
+                pid = os.fork()
+                if pid:
+                    pids.append(pid)
+                else:
+                    signal.signal(signal.SIGTERM, signal.SIG_DFL)
+                    patcher.monkey_patch(all=False, socket=True)
+                    self.object_sweep(os.path.join(self.devices, device))
+                    sys.exit()
+            else:
+                self.object_sweep(os.path.join(self.devices, device))
+        if fork:
+            while pids:
+                pids.remove(os.wait()[0])
+        elapsed = time() - begin
+        self.logger.info('Object janitor sweep completed: %.02fs' % elapsed)
+
+    def object_sweep(self, device):
+        """
+        If there are janitor pendings on the device, walk each one and update.
+
+        :param device: path to device
+        """
+        self.container_update_successes = 0
+        self.container_update_failures = 0
+        self.segment_cleanup_completions = 0
+        self.segment_cleanup_segments = 0
+        self.segment_cleanup_failures = 0
+        begin = time()
+        janitordir = os.path.join(device, JANITORDIR)
+        try:
+            if not os.path.isdir(janitordir):
+                return
+            for partition in os.listdir(janitordir):
+                partition_path = os.path.join(janitordir, partition)
+                for suffix in os.listdir(partition_path):
+                    suffix_path = os.path.join(partition_path, suffix)
+                    for janitor in os.listdir(suffix_path):
+                        janitor_path = os.path.join(suffix_path, janitor)
+                        self.process_object_janitor(device, partition,
+                                                    janitor_path)
+                        sleep(self.slowdown)
+        finally:
+            elapsed = time() - begin
+            self.logger.info('Object janitor sweep of %s completed in %.02fs: '
+                'container updates: %s successes, %s failures; segment '
+                'cleanups: %s completions, %s segments, %s failures' %
+                (device, elapsed,
+                 self.container_update_successes,
+                 self.container_update_failures,
+                 self.segment_cleanup_completions,
+                 self.segment_cleanup_segments,
+                 self.segment_cleanup_failures))
+
+    def process_object_janitor(self, device, partition, janitor_path):
+        """
+        Process the object janitor operation.
+
+        :param device: path to device
+        :param partition: partition for the object
+        :param janitor_path: path to DiskFile for the janitor operation
+        """
+        data_file = None
+        files = sorted(os.listdir(janitor_path), reverse=True)
+        for file_ in files:
+            if file_.endswith('.ts'):
+                break
+            if file_.endswith('.data'):
+                data_file = os.path.join(janitor_path, file_)
+                break
+        if not data_file:
+            return
+        metadata = read_metadata(data_file)
+        _, account, container, obj = metadata['name'].split('/', 3)
+        if metadata['X-Op'] == 'Container-Update':
+            self.process_container_update(device, partition, account,
+                                          container, obj)
+        elif metadata['X-Op'] == 'Segment-Cleanup':
+            if self.get_object_ring().get_part_nodes(int(partition))[0]['id'] \
+                    in self.my_node_ids:
+                self.process_segment_cleanup(device, partition, account,
+                                             container, obj)
+        else:
+            self.logger.error('ERROR: Unknown X-Op: %s' % metadata['X-Op'])
+            return
+
+    def process_container_update(self, device, partition, account, container,
+                                 obj):
+        """
+        Process the container update operation, sending the object information
+        to the container server.
+
+        :param device: path to device
+        :param partition: partition for the object
+        :param account: account for the object
+        :param container: container for the object
+        :param obj: name for the object
+        """
+        disk_file = DiskFile(self.devices, device, partition, account,
+                        container, obj, keep_data_fp=True, datadir=JANITORDIR)
+        if disk_file.is_deleted():
+            return
+        update = pickle.loads(''.join(iter(disk_file)))
+        op = update['op']
+        account = update['account']
+        container = update['container']
+        obj = update['obj']
+        headers = update['headers']
+        successes = update.get('successes', [])
+        part, nodes = self.get_container_ring().get_nodes(account, container)
+        path = '/%s/%s/%s' % (account, container, obj)
+        success = True
+        for node in nodes:
+            if node['id'] not in successes:
+                status = self.container_update(node, part, op, path, headers)
+                if not (200 <= status < 300) and status != 404:
+                    success = False
+                else:
+                    successes.append(node['id'])
+        if success:
+            self.container_update_successes += 1
+            self.logger.debug('Update sent for %s %s' %
+                              (path, disk_file.datadir))
+            disk_file.tombstone(normalize_timestamp(time()))
+        else:
+            self.container_update_failures += 1
+            self.logger.debug('Update failed for %s %s' %
+                              (path, disk_file.datadir))
+            if len(update.get('successes', [])) != len(successes):
+                disk_file.store_janitor_container_update(op, account,
+                    container, obj, headers, successes)
+
+    def container_update(self, node, part, op, obj, headers):
+        """
+        Perform the actual container update network operation, sending the
+        object information to the container server.
+
+        :param node: node dictionary from the container ring
+        :param part: partition that holds the container
+        :param op: operation performed (ex: 'POST' or 'DELETE')
+        :param obj: object name being updated
+        :param headers: headers to send with the update
+        """
+        try:
+            with ConnectionTimeout(self.conn_timeout):
+                conn = http_connect(node['ip'], node['port'], node['device'],
+                    part, op, obj, headers)
+            with Timeout(self.node_timeout):
+                resp = conn.getresponse()
+                resp.read()
+                return resp.status
+        except:
+            self.logger.exception('ERROR with remote server '
+                '%(ip)s:%(port)s/%(device)s' % node)
+        return 500
+
+    def process_segment_cleanup(self, device, partition, account, container,
+                                obj):
+        """
+        Process the segment cleanup operation, checking to see if the
+        operation should have completed long ago but still has no manifest and
+        therefore should have the orphaned segments removed.
+
+        :param device: path to device
+        :param partition: partition for the operation
+        :param account: account for the operation (should always be
+                        "Segment-Cleanup")
+        :param container: container for the operation (actually the timestamp
+                          of the original PUT)
+        :param obj: name for the object (actually the full path
+                    account/container/object for the original object PUT)
+        """
+        disk_file = DiskFile(self.devices, device, partition, account,
+                        container, obj, keep_data_fp=True, datadir=JANITORDIR)
+        if disk_file.is_deleted():
+            return
+        disk_file_data = pickle.loads(''.join(iter(disk_file)))
+        segment_timestamp = container
+        # If not None, indicates we have started cleaning up these segments
+        # already and should just continue the operation.
+        segment_last_deleted = disk_file_data.get('segment_last_deleted', None)
+        # Sometimes we'll have no clue how many segments there are, which is
+        # what None means.
+        segment_count = disk_file_data.get('segment_count', None)
+        account = disk_file_data['account']
+        container = disk_file_data['container']
+        obj = disk_file_data['obj']
+        path = '/%s/%s/%s' % (account, container, obj)
+        if segment_last_deleted is not None:
+            self.logger.debug('Continue cleaning up segments for %s/%s/%s %s '
+                '@%s' % (device, partition, path, segment_timestamp,
+                segment_last_deleted))
+            self.cleanup_segments(disk_file, account, container, obj,
+                segment_timestamp, segment_count, segment_last_deleted)
+            return
+        part, nodes = self.get_object_ring().get_nodes(account, container, obj)
+        newest_object = None
+        newest_manifest = None
+        responses = 0
+        for node in nodes:
+            try:
+                with ConnectionTimeout(self.conn_timeout):
+                    conn = http_connect(node['ip'], node['port'],
+                                        node['device'], part, 'GET', path)
+                with Timeout(self.node_timeout):
+                    resp = conn.getresponse()
+                responses += 1
+                if resp.status // 100 != 2:
+                    conn.close()
+                    continue
+                if resp.getheader('x-object-type', '') != 'manifest':
+                    if resp.getheader('x-timestamp') > newest_object:
+                        newest_object = resp.getheader('x-timestamp')
+                    conn.close()
+                    if newest_object > segment_timestamp:
+                        # We don't have to talk to the other nodes if we
+                        # already know we have a newer object.
+                        break
+                    continue
+                with Timeout(self.node_timeout):
+                    body = resp.read()
+                manifest = pickle.loads(body)
+                if manifest['x-timestamp'] > newest_manifest:
+                    newest_manifest = manifest['x-timestamp']
+                conn.close()
+                if newest_manifest > segment_timestamp:
+                    # We don't have to talk to the other nodes if we already
+                    # know we have a newer manifest.
+                    break
+            except:
+                self.logger.exception('ERROR with remote server '
+                    '%(ip)s:%(port)s/%(device)s' % node)
+        if newest_object > segment_timestamp or \
+                newest_manifest > segment_timestamp:
+            self.logger.debug('Newer object/manifest; discarding old '
+                'segments for %s/%s/%s %s' % (device, partition, path,
+                segment_timestamp))
+            self.cleanup_segments(disk_file, account, container, obj,
+                segment_timestamp, segment_count, segment_last_deleted)
+        elif newest_manifest == segment_timestamp:
+            self.logger.debug('Exact manifest confirmed; discarding janitor '
+                'cleanup operation for %s/%s/%s %s' % (device, partition,
+                path, segment_timestamp))
+            disk_file.tombstone(normalize_timestamp(time()))
+        elif responses == len(nodes) and \
+                time() - float(segment_timestamp) > \
+                self.segment_reclaim_age:
+            self.logger.debug('All nodes agree the manifest still does not '
+                'exist after %ss; discarding orphaned segments for '
+                '%s/%s/%s %s' % (self.segment_reclaim_age, device,
+                partition, path, segment_timestamp))
+            self.cleanup_segments(disk_file, account, container, obj,
+                segment_timestamp, segment_count, segment_last_deleted)
+
+    def cleanup_segments(self, disk_file, account, container, obj,
+            segment_timestamp, segment_count, segment_last_deleted):
+        """
+        Issues DELETEs to the segments of the object, up to
+        self.segments_per_pass. If all the segments are not deleted before this
+        function returns, it will update the disk_file with the latest point
+        completed.
+
+        :param disk_file: DiskFile for the Segment Cleanup operation.
+        :param account: Account name for the manifest object to delete segments
+                        for.
+        :param container: Container name for the manifest object to delete
+                          segments for.
+        :param obj: Object name for the manifest object to delete segments for.
+        :param segment_timestamp: The timestamp for the segments themselves.
+        :param segment_count: The number of segments for the object; may be
+                              None if not yet known and this function should
+                              try to determine the count.
+        :param segment_last_deleted: The segment number that was last deleted;
+                                     None of none have yet been deleted. This
+                                     allows continuation of previous deletion
+                                     attempts.
+        """
+        path = '/%s/%s/%s' % (account, container, obj)
+        if segment_last_deleted is None:
+            segment_last_deleted = -1
+        if segment_count is None:
+            # We need to determine how many segments there are so that if we
+            # crash this run, next run we'll know what range we'll need to work
+            # within. Example: Determine we have 10 segments; delete the first
+            # five and crash; next run we now know we need to delete no more
+            # than 10 segments. With the approach of just deleting segments
+            # until a 404 and not predetermining the count, with such a crash
+            # we'd never delete the last five segments.
+            segment_count = 0
+            assumptions = 0
+            while True:
+                if segment_count:
+                    ring_obj = \
+                        '%s/%s/%s' % (obj, segment_timestamp, segment_count)
+                else:
+                    ring_obj = obj
+                part, nodes = self.get_object_ring().get_nodes(account,
+                                container, ring_obj)
+                headers = {'X-Object-Segment': str(segment_count),
+                           'X-Object-Segment-Timestamp': segment_timestamp}
+                status = 0
+                for node in nodes:
+                    try:
+                        with ConnectionTimeout(self.conn_timeout):
+                            conn = http_connect(node['ip'], node['port'],
+                                node['device'], part, 'HEAD', path,
+                                headers=headers)
+                        with Timeout(self.node_timeout):
+                            resp = conn.getresponse()
+                            resp.read()
+                        if resp.status == 404:
+                            status = 404
+                        elif resp.status // 100 == 2:
+                            status = 200
+                            break
+                    except:
+                        self.logger.exception('ERROR with remote server '
+                            '%(ip)s:%(port)s/%(device)s' % node)
+                if status == 404:
+                    break
+                segment_count += 1
+                if not status:
+                    # We couldn't determine if a segment existed, so we'll just
+                    # assume it did and move to the next one; but we'll only do
+                    # this for five consecutive segments before just giving up
+                    # completely and trying the next run.
+                    assumptions += 1
+                    if assumptions > 5:
+                        self.segment_cleanup_failures += 1
+                        return
+                assumptions = 0
+            disk_file.store_janitor_segment_cleanup(account, container, obj,
+                segment_count, segment_last_deleted)
+        starting_segment_last_deleted = segment_last_deleted
+        try:
+            segments_this_pass = self.segments_per_pass
+            while True:
+                segment = segment_last_deleted + 1
+                if segment:
+                    ring_obj = '%s/%s/%s' % (obj, segment_timestamp, segment)
+                else:
+                    ring_obj = obj
+                part, nodes = self.get_object_ring().get_nodes(account,
+                                container, ring_obj)
+                headers = {'X-Object-Segment': str(segment),
+                           'X-Object-Segment-Timestamp': segment_timestamp,
+                           'X-Timestamp': normalize_timestamp(time())}
+                not_found_count = 0
+                success = False
+                for node in nodes:
+                    try:
+                        with ConnectionTimeout(self.conn_timeout):
+                            conn = http_connect(node['ip'], node['port'],
+                                node['device'], part, 'DELETE', path,
+                                headers=headers)
+                        with Timeout(self.node_timeout):
+                            resp = conn.getresponse()
+                            resp.read()
+                        if resp.status == 404:
+                            not_found_count += 1
+                        elif resp.status // 100 == 2:
+                            # For the sake of this clean up operation, we're
+                            # going to consider even one success complete
+                            # success.
+                            success = True
+                    except:
+                        self.logger.exception('ERROR with remote server '
+                            '%(ip)s:%(port)s/%(device)s' % node)
+                if not_found_count == len(nodes):
+                    success = True
+                if not success:
+                    # If we didn't even have one success, we'll have to leave
+                    # the janitor operation to retry later.
+                    disk_file.store_janitor_segment_cleanup(account, container,
+                        obj, segment_count, segment_last_deleted)
+                    self.segment_cleanup_failures += 1
+                    return
+                segment_last_deleted = segment
+                self.segment_cleanup_segments += 1
+                if segment >= segment_count - 1:
+                    disk_file.tombstone(normalize_timestamp(time()))
+                    self.segment_cleanup_completions += 1
+                    return
+                segments_this_pass -= 1
+                if segments_this_pass <= 0:
+                    disk_file.store_janitor_segment_cleanup(account, container,
+                        obj, segment_count, segment_last_deleted)
+                    return
+        except:
+            # If we get an unexpected exception, log it and try to update the
+            # disk file to indicate the segment last deleted.
+            self.logger.exception('ERROR unexpected exception with %s:' % path)
+            self.segment_cleanup_failures += 1
+            if segment_last_deleted != starting_segment_last_deleted:
+                disk_file.store_janitor_segment_cleanup(account, container,
+                    obj, segment_count, segment_last_deleted)

=== modified file 'swift/obj/replicator.py'
--- swift/obj/replicator.py	2010-10-19 01:05:54 +0000
+++ swift/obj/replicator.py	2010-11-08 18:51:48 +0000
@@ -19,7 +19,6 @@
 import shutil
 import time
 import logging
-import hashlib
 import itertools
 import cPickle as pickle
 
@@ -29,168 +28,16 @@
 from eventlet.support.greenlets import GreenletExit
 
 from swift.common.ring import Ring
-from swift.common.utils import whataremyips, unlink_older_than, lock_path, \
-        renamer, compute_eta, get_logger
+from swift.common.utils import compute_eta, get_logger, unlink_older_than, \
+                               whataremyips
 from swift.common.bufferedhttp import http_connect
 from swift.common.daemon import Daemon
+from swift.obj.diskfile import DATADIR, get_hashes, JANITORDIR, \
+                               recalculate_hashes, SEGMENTSDIR
+
 
 hubs.use_hub('poll')
 
-PICKLE_PROTOCOL = 2
-ONE_WEEK = 604800
-HASH_FILE = 'hashes.pkl'
-
-
-def hash_suffix(path, reclaim_age):
-    """
-    Performs reclamation and returns an md5 of all (remaining) files.
-
-    :param reclaim_age: age in seconds at which to remove tombstones
-    """
-    md5 = hashlib.md5()
-    for hsh in sorted(os.listdir(path)):
-        hsh_path = join(path, hsh)
-        files = os.listdir(hsh_path)
-        if len(files) == 1:
-            if files[0].endswith('.ts'):
-                # remove tombstones older than reclaim_age
-                ts = files[0].rsplit('.', 1)[0]
-                if (time.time() - float(ts)) > reclaim_age:
-                    os.unlink(join(hsh_path, files[0]))
-                    files.remove(files[0])
-        elif files:
-            files.sort(reverse=True)
-            meta = data = tomb = None
-            for filename in files:
-                if not meta and filename.endswith('.meta'):
-                    meta = filename
-                if not data and filename.endswith('.data'):
-                    data = filename
-                if not tomb and filename.endswith('.ts'):
-                    tomb = filename
-                if (filename < tomb or       # any file older than tomb
-                    filename < data or       # any file older than data
-                    (filename.endswith('.meta') and
-                     filename < meta)):      # old meta
-                    os.unlink(join(hsh_path, filename))
-                    files.remove(filename)
-        if not files:
-            os.rmdir(hsh_path)
-        for filename in files:
-            md5.update(filename)
-    try:
-        os.rmdir(path)
-    except OSError:
-        pass
-    return md5.hexdigest()
-
-
-def recalculate_hashes(partition_dir, suffixes, reclaim_age=ONE_WEEK):
-    """
-    Recalculates hashes for the given suffixes in the partition and updates
-    them in the partition's hashes file.
-
-    :param partition_dir: directory of the partition in which to recalculate
-    :param suffixes: list of suffixes to recalculate
-    :param reclaim_age: age in seconds at which tombstones should be removed
-    """
-
-    def tpool_listdir(partition_dir):
-        return dict(((suff, None) for suff in os.listdir(partition_dir)
-                     if len(suff) == 3 and isdir(join(partition_dir, suff))))
-    hashes_file = join(partition_dir, HASH_FILE)
-    with lock_path(partition_dir):
-        try:
-            with open(hashes_file, 'rb') as fp:
-                hashes = pickle.load(fp)
-        except Exception:
-            hashes = tpool.execute(tpool_listdir, partition_dir)
-        for suffix in suffixes:
-            suffix_dir = join(partition_dir, suffix)
-            if os.path.exists(suffix_dir):
-                hashes[suffix] = hash_suffix(suffix_dir, reclaim_age)
-            elif suffix in hashes:
-                del hashes[suffix]
-        with open(hashes_file + '.tmp', 'wb') as fp:
-            pickle.dump(hashes, fp, PICKLE_PROTOCOL)
-        renamer(hashes_file + '.tmp', hashes_file)
-
-
-def invalidate_hash(suffix_dir):
-    """
-    Invalidates the hash for a suffix_dir in the partition's hashes file.
-
-    :param suffix_dir: absolute path to suffix dir whose hash needs
-                       invalidating
-    """
-
-    suffix = os.path.basename(suffix_dir)
-    partition_dir = os.path.dirname(suffix_dir)
-    hashes_file = join(partition_dir, HASH_FILE)
-    with lock_path(partition_dir):
-        try:
-            with open(hashes_file, 'rb') as fp:
-                hashes = pickle.load(fp)
-            if suffix in hashes and not hashes[suffix]:
-                return
-        except Exception:
-            return
-        hashes[suffix] = None
-        with open(hashes_file + '.tmp', 'wb') as fp:
-            pickle.dump(hashes, fp, PICKLE_PROTOCOL)
-        renamer(hashes_file + '.tmp', hashes_file)
-
-
-def get_hashes(partition_dir, do_listdir=True, reclaim_age=ONE_WEEK):
-    """
-    Get a list of hashes for the suffix dir.  do_listdir causes it to mistrust
-    the hash cache for suffix existence at the (unexpectedly high) cost of a
-    listdir.  reclaim_age is just passed on to hash_suffix.
-
-    :param partition_dir: absolute path of partition to get hashes for
-    :param do_listdir: force existence check for all hashes in the partition
-    :param reclaim_age: age at which to remove tombstones
-
-    :returns: tuple of (number of suffix dirs hashed, dictionary of hashes)
-    """
-
-    def tpool_listdir(hashes, partition_dir):
-        return dict(((suff, hashes.get(suff, None))
-                     for suff in os.listdir(partition_dir)
-                     if len(suff) == 3 and isdir(join(partition_dir, suff))))
-    hashed = 0
-    hashes_file = join(partition_dir, HASH_FILE)
-    with lock_path(partition_dir):
-        modified = False
-        hashes = {}
-        try:
-            with open(hashes_file, 'rb') as fp:
-                hashes = pickle.load(fp)
-        except Exception:
-            do_listdir = True
-        if do_listdir:
-            hashes = tpool.execute(tpool_listdir, hashes, partition_dir)
-            modified = True
-        for suffix, hash_ in hashes.items():
-            if not hash_:
-                suffix_dir = join(partition_dir, suffix)
-                if os.path.exists(suffix_dir):
-                    try:
-                        hashes[suffix] = hash_suffix(suffix_dir, reclaim_age)
-                        hashed += 1
-                    except OSError:
-                        logging.exception('Error hashing suffix')
-                        hashes[suffix] = None
-                else:
-                    del hashes[suffix]
-                modified = True
-                sleep()
-        if modified:
-            with open(hashes_file + '.tmp', 'wb') as fp:
-                pickle.dump(hashes, fp, PICKLE_PROTOCOL)
-            renamer(hashes_file + '.tmp', hashes_file)
-        return hashed, hashes
-
 
 class ObjectReplicator(Daemon):
     """
@@ -302,7 +149,7 @@
         if not had_any:
             return False
         args.append(join(rsync_module, node['device'],
-                    'objects', job['partition']))
+                    job.get('datadir', DATADIR), job['partition']))
         return self._rsync(args) == 0
 
     def check_ring(self):
@@ -337,12 +184,14 @@
                 for node in job['nodes']:
                     success = self.rsync(node, job, suffixes)
                     if success:
+                        headers = {'Content-Length': '0',
+                                   'X-Data-Dir': job.get('datadir', DATADIR)}
                         with Timeout(self.http_timeout):
                             http_connect(node['ip'],
                                 node['port'],
                                 node['device'], job['partition'], 'REPLICATE',
                                 '/' + '-'.join(suffixes),
-                          headers={'Content-Length': '0'}).getresponse().read()
+                                headers=headers).getresponse().read()
                     responses.append(success)
             if not suffixes or (len(responses) == \
                         self.object_ring.replica_count and all(responses)):
@@ -374,10 +223,12 @@
                 node = next(nodes)
                 attempts_left -= 1
                 try:
+                    headers = {'Content-Length': '0',
+                               'X-Data-Dir': job.get('datadir', DATADIR)}
                     with Timeout(self.http_timeout):
                         resp = http_connect(node['ip'], node['port'],
                                 node['device'], job['partition'], 'REPLICATE',
-                            '', headers={'Content-Length': '0'}).getresponse()
+                                '', headers=headers).getresponse()
                         if resp.status == 507:
                             self.logger.error('%s/%s responded as unmounted' %
                                               (node['ip'], node['device']))
@@ -397,11 +248,13 @@
                     self.rsync(node, job, suffixes)
                     recalculate_hashes(job['path'], suffixes,
                                        reclaim_age=self.reclaim_age)
+                    headers = {'Content-Length': '0',
+                               'X-Data-Dir': job.get('datadir', DATADIR)}
                     with Timeout(self.http_timeout):
                         conn = http_connect(node['ip'], node['port'],
                             node['device'], job['partition'], 'REPLICATE',
                             '/' + '-'.join(suffixes),
-                            headers={'Content-Length': '0'})
+                            headers=headers)
                         conn.getresponse().read()
                     self.suffix_sync += len(suffixes)
                 except (Exception, Timeout):
@@ -489,24 +342,27 @@
                     dev for dev in self.object_ring.devs
                     if dev and dev['ip'] in ips and dev['port'] == self.port]:
                 dev_path = join(self.devices_dir, local_dev['device'])
-                obj_path = join(dev_path, 'objects')
-                tmp_path = join(dev_path, 'tmp')
                 if self.mount_check and not os.path.ismount(dev_path):
                     self.logger.warn('%s is not mounted' % local_dev['device'])
                     continue
+                tmp_path = join(dev_path, 'tmp')
                 unlink_older_than(tmp_path, time.time() - self.reclaim_age)
-                if not os.path.exists(obj_path):
-                    continue
-                for partition in os.listdir(obj_path):
-                    try:
-                        nodes = [node for node in
-                            self.object_ring.get_part_nodes(int(partition))
-                                 if node['id'] != local_dev['id']]
-                        jobs.append(dict(path=join(obj_path, partition),
-                            nodes=nodes, delete=len(nodes) > 2,
-                            partition=partition))
-                    except ValueError:
-                        continue
+                for datadir in (DATADIR, JANITORDIR, SEGMENTSDIR):
+                    obj_path = join(dev_path, datadir)
+                    if os.path.exists(obj_path):
+                        for partition in os.listdir(obj_path):
+                            try:
+                                nodes = [node for node in
+                                    self.object_ring.get_part_nodes(
+                                        int(partition))
+                                    if node['id'] != local_dev['id']]
+                                jobs.append(dict(
+                                    path=join(obj_path, partition),
+                                    nodes=nodes, delete=len(nodes) > 2,
+                                    partition=partition,
+                                    datadir=datadir))
+                            except ValueError:
+                                continue
             random.shuffle(jobs)
             # Partititons that need to be deleted take priority
             jobs.sort(key=lambda job: not job['delete'])

=== modified file 'swift/obj/server.py'
--- swift/obj/server.py	2010-11-01 21:47:48 +0000
+++ swift/obj/server.py	2010-11-08 18:51:48 +0000
@@ -17,225 +17,28 @@
 
 from __future__ import with_statement
 import cPickle as pickle
-import errno
 import os
 import time
 import traceback
 from datetime import datetime
 from hashlib import md5
-from tempfile import mkstemp
 from urllib import unquote
-from contextlib import contextmanager
 
 from webob import Request, Response, UTC
 from webob.exc import HTTPAccepted, HTTPBadRequest, HTTPCreated, \
     HTTPInternalServerError, HTTPNoContent, HTTPNotFound, \
     HTTPNotModified, HTTPPreconditionFailed, \
     HTTPRequestTimeout, HTTPUnprocessableEntity, HTTPMethodNotAllowed
-from xattr import getxattr, setxattr
 from eventlet import sleep, Timeout
 
-from swift.common.utils import mkdirs, normalize_timestamp, \
-    storage_directory, hash_path, renamer, fallocate, \
-    split_path, drop_buffer_cache, get_logger, write_pickle
+from swift.common.utils import drop_buffer_cache, fallocate, get_logger, \
+                               mkdirs, normalize_timestamp, split_path
 from swift.common.bufferedhttp import http_connect
 from swift.common.constraints import check_object_creation, check_mount, \
     check_float, check_utf8
 from swift.common.exceptions import ConnectionTimeout
-from swift.obj.replicator import get_hashes, invalidate_hash, \
-    recalculate_hashes
-
-
-DATADIR = 'objects'
-ASYNCDIR = 'async_pending'
-PICKLE_PROTOCOL = 2
-METADATA_KEY = 'user.swift.metadata'
-MAX_OBJECT_NAME_LENGTH = 1024
-
-
-def read_metadata(fd):
-    """
-    Helper function to read the pickled metadata from an object file.
-
-    :param fd: file descriptor to load the metadata from
-
-    :returns: dictionary of metadata
-    """
-    metadata = ''
-    key = 0
-    try:
-        while True:
-            metadata += getxattr(fd, '%s%s' % (METADATA_KEY, (key or '')))
-            key += 1
-    except IOError:
-        pass
-    return pickle.loads(metadata)
-
-
-class DiskFile(object):
-    """
-    Manage object files on disk.
-
-    :param path: path to devices on the node
-    :param device: device name
-    :param partition: partition on the device the object lives in
-    :param account: account name for the object
-    :param container: container name for the object
-    :param obj: object name for the object
-    :param keep_data_fp: if True, don't close the fp, otherwise close it
-    :param disk_chunk_Size: size of chunks on file reads
-    """
-
-    def __init__(self, path, device, partition, account, container, obj,
-                    keep_data_fp=False, disk_chunk_size=65536):
-        self.disk_chunk_size = disk_chunk_size
-        self.name = '/' + '/'.join((account, container, obj))
-        name_hash = hash_path(account, container, obj)
-        self.datadir = os.path.join(path, device,
-                    storage_directory(DATADIR, partition, name_hash))
-        self.tmpdir = os.path.join(path, device, 'tmp')
-        self.metadata = {}
-        self.meta_file = None
-        self.data_file = None
-        if not os.path.exists(self.datadir):
-            return
-        files = sorted(os.listdir(self.datadir), reverse=True)
-        for file in files:
-            if file.endswith('.ts'):
-                self.data_file = self.meta_file = None
-                self.metadata = {'deleted': True}
-                return
-            if file.endswith('.meta') and not self.meta_file:
-                self.meta_file = os.path.join(self.datadir, file)
-            if file.endswith('.data') and not self.data_file:
-                self.data_file = os.path.join(self.datadir, file)
-                break
-        if not self.data_file:
-            return
-        self.fp = open(self.data_file, 'rb')
-        self.metadata = read_metadata(self.fp)
-        if not keep_data_fp:
-            self.close()
-        if self.meta_file:
-            with open(self.meta_file) as mfp:
-                for key in self.metadata.keys():
-                    if key.lower() not in ('content-type', 'content-encoding',
-                                'deleted', 'content-length', 'etag'):
-                        del self.metadata[key]
-                self.metadata.update(read_metadata(mfp))
-
-    def __iter__(self):
-        """Returns an iterator over the data file."""
-        try:
-            dropped_cache = 0
-            read = 0
-            while True:
-                chunk = self.fp.read(self.disk_chunk_size)
-                if chunk:
-                    read += len(chunk)
-                    if read - dropped_cache > (1024 * 1024):
-                        drop_buffer_cache(self.fp.fileno(), dropped_cache,
-                            read - dropped_cache)
-                        dropped_cache = read
-                    yield chunk
-                else:
-                    drop_buffer_cache(self.fp.fileno(), dropped_cache,
-                        read - dropped_cache)
-                    break
-        finally:
-            self.close()
-
-    def app_iter_range(self, start, stop):
-        """Returns an iterator over the data file for range (start, stop)"""
-        if start:
-            self.fp.seek(start)
-        if stop is not None:
-            length = stop - start
-        else:
-            length = None
-        for chunk in self:
-            if length is not None:
-                length -= len(chunk)
-                if length < 0:
-                    # Chop off the extra:
-                    yield chunk[:length]
-                    break
-            yield chunk
-
-    def close(self):
-        """Close the file."""
-        if self.fp:
-            self.fp.close()
-            self.fp = None
-
-    def is_deleted(self):
-        """
-        Check if the file is deleted.
-
-        :returns: True if the file doesn't exist or has been flagged as
-                  deleted.
-        """
-        return not self.data_file or 'deleted' in self.metadata
-
-    @contextmanager
-    def mkstemp(self):
-        """Contextmanager to make a temporary file."""
-        if not os.path.exists(self.tmpdir):
-            mkdirs(self.tmpdir)
-        fd, tmppath = mkstemp(dir=self.tmpdir)
-        try:
-            yield fd, tmppath
-        finally:
-            try:
-                os.close(fd)
-            except OSError:
-                pass
-            try:
-                os.unlink(tmppath)
-            except OSError:
-                pass
-
-    def put(self, fd, tmppath, metadata, extension='.data'):
-        """
-        Finalize writing the file on disk, and renames it from the temp file to
-        the real location.  This should be called after the data has been
-        written to the temp file.
-
-        :params fd: file descriptor of the temp file
-        :param tmppath: path to the temporary file being used
-        :param metadata: dictionary of metada to be written
-        :param extention: extension to be used when making the file
-        """
-        metadata['name'] = self.name
-        timestamp = normalize_timestamp(metadata['X-Timestamp'])
-        metastr = pickle.dumps(metadata, PICKLE_PROTOCOL)
-        key = 0
-        while metastr:
-            setxattr(fd, '%s%s' % (METADATA_KEY, key or ''), metastr[:254])
-            metastr = metastr[254:]
-            key += 1
-        if 'Content-Length' in metadata:
-            drop_buffer_cache(fd, 0, int(metadata['Content-Length']))
-        os.fsync(fd)
-        invalidate_hash(os.path.dirname(self.datadir))
-        renamer(tmppath, os.path.join(self.datadir, timestamp + extension))
-        self.metadata = metadata
-
-    def unlinkold(self, timestamp):
-        """
-        Remove any older versions of the object file.  Any file that has an
-        older timestamp than timestamp will be deleted.
-
-        :param timestamp: timestamp to compare with each file
-        """
-        timestamp = normalize_timestamp(timestamp)
-        for fname in os.listdir(self.datadir):
-            if fname < timestamp:
-                try:
-                    os.unlink(os.path.join(self.datadir, fname))
-                except OSError, err:    # pragma: no cover
-                    if err.errno != errno.ENOENT:
-                        raise
+from swift.obj.diskfile import DATADIR, DiskFile, get_hashes, JANITORDIR, \
+                               recalculate_hashes
 
 
 class ObjectController(object):
@@ -262,7 +65,7 @@
         self.bytes_per_sync = int(conf.get('mb_per_sync', 512)) * 1024 * 1024
 
     def container_update(self, op, account, container, obj, headers_in,
-                         headers_out, objdevice):
+                         headers_out, objdevice, objpartition):
         """
         Update the container when objects are updated.
 
@@ -274,6 +77,7 @@
         :param headers_out: dictionary of headers to send in the container
                             request
         :param objdevice: device name that the object is in
+        :param objpartition: partition that the object is in
         """
         host = headers_in.get('X-Container-Host', None)
         partition = headers_in.get('X-Container-Partition', None)
@@ -293,20 +97,16 @@
                     return
                 else:
                     self.logger.error('ERROR Container update failed (saving '
-                        'for async update later): %d response from %s:%s/%s' %
-                        (response.status, ip, port, contdevice))
+                        'for janitor update later): %d response from %s:%s/%s'
+                        % (response.status, ip, port, contdevice))
         except:
             self.logger.exception('ERROR container update failed with '
-                '%s:%s/%s transaction %s (saving for async update later)' %
+                '%s:%s/%s transaction %s (saving for janitor update later)' %
                 (ip, port, contdevice, headers_in.get('x-cf-trans-id', '-')))
-        async_dir = os.path.join(self.devices, objdevice, ASYNCDIR)
-        ohash = hash_path(account, container, obj)
-        write_pickle(
-            {'op': op, 'account': account, 'container': container,
-                'obj': obj, 'headers': headers_out},
-            os.path.join(async_dir, ohash[-3:], ohash + '-' +
-                normalize_timestamp(headers_out['x-timestamp'])),
-            os.path.join(self.devices, objdevice, 'tmp'))
+        df = DiskFile(self.devices, objdevice, objpartition, account,
+                      container, obj, datadir=JANITORDIR)
+        df.store_janitor_container_update(op, account, container, obj,
+                                          headers_out, [])
 
     def POST(self, request):
         """Handle HTTP POST requests for the Swift Object Server."""
@@ -355,7 +155,15 @@
         if error_response:
             return error_response
         file = DiskFile(self.devices, device, partition, account, container,
-                        obj, disk_chunk_size=self.disk_chunk_size)
+                obj, disk_chunk_size=self.disk_chunk_size, keep_data_fp=True,
+                segment=request.headers.get('x-object-segment'),
+                segment_timestamp=request.headers['x-timestamp'])
+        overwritten_manifest = False
+        if not file.is_deleted() and \
+                file.metadata.get('X-Object-Type') == 'manifest' and \
+                'x-object-segment' not in request.headers:
+            overwritten_manifest = pickle.loads(''.join(iter(file)))
+        file.close()
         upload_expiration = time.time() + self.max_upload_time
         etag = md5()
         upload_size = 0
@@ -397,17 +205,51 @@
             if 'content-encoding' in request.headers:
                 metadata['Content-Encoding'] = \
                     request.headers['Content-Encoding']
-            file.put(fd, tmppath, metadata)
+            if 'x-object-type' in request.headers:
+                metadata['X-Object-Type'] = request.headers['x-object-type']
+            if 'x-object-segment' in request.headers:
+                metadata['X-Object-Segment'] = \
+                    request.headers['x-object-segment']
+            no_longer_segment = False
+            if 'x-object-segment-if-length' in request.headers and \
+                    int(request.headers['x-object-segment-if-length']) != \
+                    os.fstat(fd).st_size:
+                del metadata['X-Object-Type']
+                del metadata['X-Object-Segment']
+                no_longer_segment = True
+            elif int(request.headers.get('x-object-segment', -1)) == 0:
+                # Write out a janitor operation to clean up this multi-segment
+                # PUT in the future if it fails.
+                df = DiskFile(self.devices, device, partition,
+                        'Segment-Cleanup', request.headers['x-timestamp'],
+                        '%s/%s/%s' % (account, container, obj),
+                        datadir=JANITORDIR)
+                df.store_janitor_segment_cleanup(account, container, obj,
+                    segment_count=None, segment_last_deleted=None)
+            if overwritten_manifest:
+                # Write out a janitor operation to clean up the overwritten
+                # multi-segment object.
+                df = DiskFile(self.devices, device, partition,
+                    'Segment-Cleanup', overwritten_manifest['x-timestamp'],
+                    '%s/%s/%s' % (account, container, obj), datadir=JANITORDIR)
+                df.store_janitor_segment_cleanup(account, container, obj,
+                    segment_count=(overwritten_manifest['content-length'] /
+                                   overwritten_manifest['x-segment-size'] + 1),
+                    segment_last_deleted=None)
+            file.put(fd, tmppath, metadata,
+                     no_longer_segment=no_longer_segment)
         file.unlinkold(metadata['X-Timestamp'])
-        self.container_update('PUT', account, container, obj, request.headers,
-            {'x-size': file.metadata['Content-Length'],
-             'x-content-type': file.metadata['Content-Type'],
-             'x-timestamp': file.metadata['X-Timestamp'],
-             'x-etag': file.metadata['ETag'],
-             'x-cf-trans-id': request.headers.get('x-cf-trans-id', '-')},
-            device)
-        resp = HTTPCreated(request=request, etag=etag)
-        return resp
+        if 'X-Object-Segment' not in file.metadata:
+            self.container_update('PUT', account, container, obj,
+                request.headers,
+                {'x-size': request.headers.get('x-object-length',
+                                              file.metadata['Content-Length']),
+                 'x-content-type': file.metadata['Content-Type'],
+                 'x-timestamp': file.metadata['X-Timestamp'],
+                 'x-etag': file.metadata['ETag'],
+                 'x-cf-trans-id': request.headers.get('x-cf-trans-id', '-')},
+                device, partition)
+        return HTTPCreated(request=request, etag=etag)
 
     def GET(self, request):
         """Handle HTTP GET requests for the Swift Object Server."""
@@ -420,7 +262,9 @@
         if self.mount_check and not check_mount(self.devices, device):
             return Response(status='507 %s is not mounted' % device)
         file = DiskFile(self.devices, device, partition, account, container,
-                obj, keep_data_fp=True, disk_chunk_size=self.disk_chunk_size)
+           obj, keep_data_fp=True, disk_chunk_size=self.disk_chunk_size,
+           segment=request.headers.get('x-object-segment'),
+           segment_timestamp=request.headers.get('x-object-segment-timestamp'))
         if file.is_deleted():
             if request.headers.get('if-match') == '*':
                 return HTTPPreconditionFailed(request=request)
@@ -460,7 +304,9 @@
                         'application/octet-stream'), app_iter=file,
                         request=request, conditional_response=True)
         for key, value in file.metadata.iteritems():
-            if key.lower().startswith('x-object-meta-'):
+            if key.lower().startswith('x-object-meta-') or \
+                    key.lower() in ('x-timestamp', 'x-object-type',
+                                    'x-object-segment'):
                 response.headers[key] = value
         response.etag = file.metadata['ETag']
         response.last_modified = float(file.metadata['X-Timestamp'])
@@ -482,13 +328,17 @@
         if self.mount_check and not check_mount(self.devices, device):
             return Response(status='507 %s is not mounted' % device)
         file = DiskFile(self.devices, device, partition, account, container,
-                        obj, disk_chunk_size=self.disk_chunk_size)
+           obj, disk_chunk_size=self.disk_chunk_size,
+           segment=request.headers.get('x-object-segment'),
+           segment_timestamp=request.headers.get('x-object-segment-timestamp'))
         if file.is_deleted():
             return HTTPNotFound(request=request)
         response = Response(content_type=file.metadata['Content-Type'],
                             request=request, conditional_response=True)
         for key, value in file.metadata.iteritems():
-            if key.lower().startswith('x-object-meta-'):
+            if key.lower().startswith('x-object-meta-') or \
+                    key.lower() in ('x-timestamp', 'x-object-type',
+                                    'x-object-segment'):
                 response.headers[key] = value
         response.etag = file.metadata['ETag']
         response.last_modified = float(file.metadata['X-Timestamp'])
@@ -513,21 +363,43 @@
             return Response(status='507 %s is not mounted' % device)
         response_class = HTTPNoContent
         file = DiskFile(self.devices, device, partition, account, container,
-                        obj, disk_chunk_size=self.disk_chunk_size)
+           obj, disk_chunk_size=self.disk_chunk_size, keep_data_fp=True,
+           segment=request.headers.get('x-object-segment'),
+           segment_timestamp=request.headers.get('x-object-segment-timestamp'))
+        deleted_manifest = False
         if file.is_deleted():
             response_class = HTTPNotFound
+        elif 'x-object-segment' not in request.headers and \
+                file.metadata.get('X-Object-Type') == 'manifest':
+            deleted_manifest = pickle.loads(''.join(iter(file)))
+        file.close()
         metadata = {
             'X-Timestamp': request.headers['X-Timestamp'], 'deleted': True,
         }
         with file.mkstemp() as (fd, tmppath):
+            if deleted_manifest:
+                # Write out a janitor operation to clean up the deleted
+                # multi-segment object. Note that setting the
+                # segment_last_deleted = -1 will cause the object-janitor to
+                # start removing the segments immediately rather than waiting
+                # segment_reclaim_age (otherwise it can't tell the difference
+                # between a deleted manifest and manifest that just hasn't
+                # appeared yet).
+                df = DiskFile(self.devices, device, partition,
+                    'Segment-Cleanup', deleted_manifest['x-timestamp'],
+                    '%s/%s/%s' % (account, container, obj), datadir=JANITORDIR)
+                df.store_janitor_segment_cleanup(account, container, obj,
+                    segment_count=(deleted_manifest['content-length'] /
+                                   deleted_manifest['x-segment-size']) + 1,
+                    segment_last_deleted=-1)
             file.put(fd, tmppath, metadata, extension='.ts')
         file.unlinkold(metadata['X-Timestamp'])
-        self.container_update('DELETE', account, container, obj,
-            request.headers, {'x-timestamp': metadata['X-Timestamp'],
-            'x-cf-trans-id': request.headers.get('x-cf-trans-id', '-')},
-            device)
-        resp = response_class(request=request)
-        return resp
+        if 'x-object-segment' not in request.headers:
+            self.container_update('DELETE', account, container, obj,
+                request.headers, {'x-timestamp': metadata['X-Timestamp'],
+                'x-cf-trans-id': request.headers.get('x-cf-trans-id', '-')},
+                device, partition)
+        return response_class(request=request)
 
     def REPLICATE(self, request):
         """
@@ -542,7 +414,8 @@
                                   content_type='text/plain')
         if self.mount_check and not check_mount(self.devices, device):
             return Response(status='507 %s is not mounted' % device)
-        path = os.path.join(self.devices, device, DATADIR, partition)
+        path = os.path.join(self.devices, device,
+                request.headers.get('x-data-dir', DATADIR), partition)
         if not os.path.exists(path):
             mkdirs(path)
         if suffix:

=== modified file 'swift/obj/updater.py'
--- swift/obj/updater.py	2010-09-23 16:09:30 +0000
+++ swift/obj/updater.py	2010-11-08 18:51:48 +0000
@@ -27,11 +27,25 @@
 from swift.common.ring import Ring
 from swift.common.utils import get_logger, renamer, write_pickle
 from swift.common.daemon import Daemon
-from swift.obj.server import ASYNCDIR
+
+
+# Old-style async pending directory
+ASYNCDIR = 'async_pending'
 
 
 class ObjectUpdater(Daemon):
-    """Update object information in container listings."""
+    """
+    Update object information in container listings based on postponed
+    operations stored in the old-style async pending directory.
+
+    After upgrade, no new operations will be stored in this old-style async
+    pending directory. Once this daemon empties those directories of all
+    operations, the daemon may be disabled and the directories removed.
+
+    In a future release of Swift, this daemon will be removed.
+
+    The new functionality is in swift.obj.janitor.
+    """
 
     def __init__(self, conf):
         self.conf = conf

=== modified file 'swift/proxy/server.py'
--- swift/proxy/server.py	2010-11-05 14:47:43 +0000
+++ swift/proxy/server.py	2010-11-08 18:51:48 +0000
@@ -14,21 +14,22 @@
 # limitations under the License.
 
 from __future__ import with_statement
+import cPickle as pickle
 import mimetypes
 import os
 import time
 import traceback
 from ConfigParser import ConfigParser
+from hashlib import md5
 from urllib import unquote, quote
 import uuid
 import functools
 
 from eventlet.timeout import Timeout
-from webob.exc import HTTPBadRequest, HTTPMethodNotAllowed, \
-    HTTPNotFound, HTTPPreconditionFailed, \
-    HTTPRequestTimeout, HTTPServiceUnavailable, \
-    HTTPUnprocessableEntity, HTTPRequestEntityTooLarge, HTTPServerError, \
-    status_map
+from webob.exc import HTTPBadRequest, HTTPCreated, HTTPInternalServerError, \
+    HTTPMethodNotAllowed, HTTPNotFound, HTTPPreconditionFailed, \
+    HTTPRequestEntityTooLarge, HTTPRequestTimeout, HTTPServerError, \
+    HTTPServiceUnavailable, HTTPUnprocessableEntity, status_map
 from webob import Request, Response
 
 from swift.common.ring import Ring
@@ -37,7 +38,7 @@
 from swift.common.bufferedhttp import http_connect
 from swift.common.constraints import check_metadata, check_object_creation, \
     check_utf8, MAX_ACCOUNT_NAME_LENGTH, MAX_CONTAINER_NAME_LENGTH, \
-    MAX_FILE_SIZE
+    PICKLE_PROTOCOL
 from swift.common.exceptions import ChunkReadTimeout, \
     ChunkWriteTimeout, ConnectionTimeout
 
@@ -89,6 +90,144 @@
     return wrapped
 
 
+class SegmentedIterable(object):
+    """
+    Iterable that returns the object contents for a segmented object in Swift.
+
+    In addition to these params, you can also set the `response` attr just
+    after creating the SegmentedIterable and it will update the response's
+    `bytes_transferred` value (used to log the size of the request).
+
+    :param controller: The ObjectController instance to work with.
+    :param content_length: The total length of the object.
+    :param segment_size: The length of each segment (except perhaps the last)
+                         of the object.
+    :param timestamp: The X-Timestamp of the object's segments (set on the PUT,
+                      not changed on the POSTs).
+    """
+
+    def __init__(self, controller, content_length, segment_size, timestamp):
+        self.controller = controller
+        self.content_length = content_length
+        self.segment_size = segment_size
+        self.timestamp = timestamp
+        self.position = 0
+        self.segment = -1
+        self.segment_iter = None
+        self.response = None
+
+    def load_next_segment(self):
+        """ Loads the self.segment_iter with the next segment's contents. """
+        self.segment += 1
+        if self.segment:
+            ring_object_name = '%s/%s/%s' % (self.controller.object_name,
+                self.timestamp, self.segment)
+        else:
+            ring_object_name = self.controller.object_name
+        partition, nodes = self.controller.app.object_ring.get_nodes(
+            self.controller.account_name, self.controller.container_name,
+            ring_object_name)
+        path = '/%s/%s/%s' % (self.controller.account_name,
+            self.controller.container_name, self.controller.object_name)
+        req = Request.blank(path, headers={'X-Object-Segment': self.segment,
+                'X-Object-Segment-Timestamp': self.timestamp})
+        resp = self.controller.GETorHEAD_base(req, 'Object',
+            partition, self.controller.iter_nodes(partition, nodes,
+            self.controller.app.object_ring), path,
+            self.controller.app.object_ring.replica_count)
+        if resp.status_int // 100 != 2:
+            raise Exception(
+                'Could not load segment %s of %s' % (self.segment, path))
+        self.segment_iter = resp.app_iter
+
+    def __iter__(self):
+        """ Standard iterator function that returns the object's contents. """
+        while self.position < self.content_length:
+            if not self.segment_iter:
+                self.load_next_segment()
+            while True:
+                with ChunkReadTimeout(self.controller.app.node_timeout):
+                    try:
+                        chunk = self.segment_iter.next()
+                        break
+                    except StopIteration:
+                        self.load_next_segment()
+            if self.position + len(chunk) > self.content_length:
+                chunk = chunk[:self.content_length - self.position]
+            self.position += len(chunk)
+            if self.response:
+                self.response.bytes_transferred = \
+                    getattr(self.response, 'bytes_transferred', 0) + len(chunk)
+            yield chunk
+
+    def app_iter_range(self, start, stop):
+        """
+        Non-standard iterator function for use with Webob in serving Range
+        requests more quickly.
+
+        .. note::
+
+            This currently helps on speed by jumping to the proper segment to
+            start with (and ending without reading the trailing segments, but
+            that already happened technically with __iter__).
+
+            But, what it does not do yet is issue a Range request with the
+            first segment to allow the object server to seek to the segment
+            start point.
+
+            Instead, it just reads and throws away all leading segment data.
+            Since segments are 2G by default, it'll have to transfer the whole
+            2G from the object server to the proxy server even if it only needs
+            the last byte. In practice, this should happen fairly quickly
+            relative to how long requests take for these very large files; but
+            it's still wasteful.
+
+            Anyway, it shouldn't be too hard to implement, I just want to keep
+            the complexity down for now.
+
+        :param start: The first byte (zero-based) to return. None for 0.
+        :param stop: The last byte (zero-based) to return. None for end.
+        """
+        if start is None:
+            start = 0
+        if start:
+            self.segment = (start / self.segment_size) - 1
+            self.load_next_segment()
+            self.position = self.segment * self.segment_size
+            segment_start = start - (self.segment * self.segment_size)
+            while segment_start:
+                with ChunkReadTimeout(self.controller.app.node_timeout):
+                    chunk = self.segment_iter.next()
+                    self.position += len(chunk)
+                if len(chunk) > segment_start:
+                    chunk = chunk[segment_start:]
+                    if self.response:
+                        self.response.bytes_transferred = \
+                            getattr(self.response, 'bytes_transferred', 0) + \
+                            len(chunk)
+                    yield chunk
+                    segment_start = 0
+                else:
+                    segment_start -= len(chunk)
+        if stop is not None:
+            length = stop - start
+        else:
+            length = None
+        for chunk in self:
+            if length is not None:
+                length -= len(chunk)
+                if length < 0:
+                    # bytes_transferred had len(chunk) added by __iter__ so we
+                    # need to subtract what we aren't going to use of the chunk
+                    if self.response:
+                        self.response.bytes_transferred = \
+                            getattr(self.response, 'bytes_transferred',
+                                    length) + length
+                    yield chunk[:length]
+                    break
+            yield chunk
+
+
 def get_container_memcache_key(account, container):
     path = '/%s/%s' % (account, container)
     return 'container%s' % path
@@ -518,11 +657,56 @@
             aresp = req.environ['swift.authorize'](req)
             if aresp:
                 return aresp
+        # This is bit confusing, so an explanation:
+        # * First we attempt the GET/HEAD normally, as this is the usual case.
+        # * If the request was a Range request and gave us a 416 Unsatisfiable
+        #   response, we might be trying to do an invalid Range on a manifest
+        #   object, so we try again with no Range.
+        # * If it turns out we have a manifest object, and we had a Range
+        #   request originally that actually succeeded or we had a HEAD
+        #   request, we have to do the request again as a full GET because
+        #   we'll need the whole manifest.
+        # * Finally, if we had a manifest object, we pass it and the request
+        #   off to GETorHEAD_segmented; otherwise we just return the response.
         partition, nodes = self.app.object_ring.get_nodes(
             self.account_name, self.container_name, self.object_name)
-        return self.GETorHEAD_base(req, 'Object', partition,
+        resp = mresp = self.GETorHEAD_base(req, 'Object', partition,
+            self.iter_nodes(partition, nodes, self.app.object_ring),
+            req.path_info, self.app.object_ring.replica_count)
+        range_value = None
+        if mresp.status_int == 416:
+            range_value = req.range
+            req.range = None
+            mresp = self.GETorHEAD_base(req, 'Object', partition,
                 self.iter_nodes(partition, nodes, self.app.object_ring),
                 req.path_info, self.app.object_ring.replica_count)
+        if mresp.status_int // 100 != 2:
+            return resp
+        if 'x-object-type' in mresp.headers:
+            if mresp.headers['x-object-type'] == 'manifest':
+                if req.method == 'HEAD':
+                    req.method = 'GET'
+                    mresp = self.GETorHEAD_base(req, 'Object', partition,
+                        self.iter_nodes(partition, nodes,
+                        self.app.object_ring), req.path_info,
+                        self.app.object_ring.replica_count)
+                    if mresp.status_int // 100 != 2:
+                        return mresp
+                    req.method = 'HEAD'
+                elif req.range:
+                    range_value = req.range
+                    req.range = None
+                    mresp = self.GETorHEAD_base(req, 'Object', partition,
+                        self.iter_nodes(partition, nodes,
+                        self.app.object_ring), req.path_info,
+                        self.app.object_ring.replica_count)
+                    if mresp.status_int // 100 != 2:
+                        return mresp
+                if range_value:
+                    req.range = range_value
+                return self.GETorHEAD_segmented(req, mresp)
+            return HTTPNotFound(request=req)
+        return resp
 
     @public
     @delay_denial
@@ -536,6 +720,32 @@
         """Handler for HTTP HEAD requests."""
         return self.GETorHEAD(req)
 
+    def GETorHEAD_segmented(self, req, mresp):
+        """
+        Performs a GET for a segmented object.
+
+        :param req: The webob.Request to process.
+        :param mresp: The webob.Response for the original manifest request.
+        :returns: webob.Response object.
+        """
+        manifest = pickle.loads(''.join(mresp.app_iter))
+        content_length = int(manifest['content-length'])
+        segment_size = int(manifest['x-segment-size'])
+        headers = dict(mresp.headers)
+        headers.update(manifest)
+        del headers['x-segment-size']
+        resp = Response(app_iter=SegmentedIterable(self, content_length,
+            segment_size, manifest['x-timestamp']), headers=headers,
+            request=req, conditional_response=True)
+        resp.headers['etag'] = manifest['etag'].strip('"')
+        resp.last_modified = mresp.last_modified
+        resp.content_length = int(manifest['content-length'])
+        resp.content_type = manifest['content-type']
+        if 'content-encoding' in manifest:
+            resp.content_encoding = manifest['content-encoding']
+        resp.app_iter.response = req.get_response(resp)
+        return resp.app_iter.response
+
     @public
     @delay_denial
     def POST(self, req):
@@ -609,7 +819,8 @@
                 req.headers['Content-Type'] = 'application/octet-stream'
             else:
                 req.headers['Content-Type'] = guessed_type
-        error_response = check_object_creation(req, self.object_name)
+        error_response = check_object_creation(req, self.object_name,
+                                               self.app.max_object_size)
         if error_response:
             return error_response
         conns = []
@@ -654,11 +865,50 @@
                 if k.lower().startswith('x-object-meta-'):
                     new_req.headers[k] = v
             req = new_req
+        if req.headers.get('transfer-encoding') == 'chunked' or \
+                req.content_length > self.app.segment_size:
+            resp = self.PUT_segmented_object(req, data_source, partition,
+                nodes, container_partition, containers)
+        else:
+            resp = self.PUT_whole_object(req, data_source, partition, nodes,
+                container_partition, containers)
+        if source_header:
+            resp.headers['X-Copied-From'] = quote(
+                                                source_header.split('/', 2)[2])
+            for k, v in req.headers.items():
+                if k.lower().startswith('x-object-meta-'):
+                    resp.headers[k] = v
+            # reset the bytes, since the user didn't actually send anything
+            req.bytes_transferred = 0
+        resp.last_modified = float(req.headers['X-Timestamp'])
+        return resp
+
+    def PUT_whole_object(self, req, data_source, partition, nodes,
+                         container_partition=None, containers=None):
+        """
+        Performs a PUT for a whole object (one with a content-length <=
+        self.app.segment_size).
+
+        :param req: The webob.Request to process.
+        :param data_source: An iterator providing the data to store.
+        :param partition: The object ring partition the object falls on.
+        :param nodes: The object ring nodes the object falls on.
+        :param container_partition: The container ring partition the container
+                                    for the object falls on, None if the
+                                    container is not to be updated.
+        :param containers: The container ring nodes the container for the
+                           object falls on, None if the container is not to be
+                           updated.
+        :returns: webob.Response object.
+        """
+        conns = []
+        update_containers = containers is not None
         for node in self.iter_nodes(partition, nodes, self.app.object_ring):
-            container = containers.pop()
-            req.headers['X-Container-Host'] = '%(ip)s:%(port)s' % container
-            req.headers['X-Container-Partition'] = container_partition
-            req.headers['X-Container-Device'] = container['device']
+            if update_containers:
+                container = containers.pop()
+                req.headers['X-Container-Host'] = '%(ip)s:%(port)s' % container
+                req.headers['X-Container-Partition'] = container_partition
+                req.headers['X-Container-Device'] = container['device']
             req.headers['Expect'] = '100-continue'
             resp = conn = None
             if not self.error_limited(node):
@@ -676,12 +926,14 @@
             if conn and resp:
                 if resp.status == 100:
                     conns.append(conn)
-                    if not containers:
+                    if (update_containers and not containers) or \
+                            len(conns) == len(nodes):
                         break
                     continue
                 elif resp.status == 507:
                     self.error_limit(node)
-            containers.insert(0, container)
+            if update_containers:
+                containers.insert(0, container)
         if len(conns) <= len(nodes) / 2:
             self.app.logger.error(
                 'Object PUT returning 503, %s/%s required connections, '
@@ -701,7 +953,7 @@
                             break
                 len_chunk = len(chunk)
                 req.bytes_transferred += len_chunk
-                if req.bytes_transferred > MAX_FILE_SIZE:
+                if req.bytes_transferred > self.app.max_object_size:
                     return HTTPRequestEntityTooLarge(request=req)
                 for conn in list(conns):
                     try:
@@ -767,18 +1019,129 @@
             statuses.append(503)
             reasons.append('')
             bodies.append('')
-        resp = self.best_response(req, statuses, reasons, bodies, 'Object PUT',
+        return self.best_response(req, statuses, reasons, bodies, 'Object PUT',
                                   etag=etag)
-        if source_header:
-            resp.headers['X-Copied-From'] = quote(
-                                                source_header.split('/', 2)[2])
-            for k, v in req.headers.items():
-                if k.lower().startswith('x-object-meta-'):
-                    resp.headers[k] = v
-            # reset the bytes, since the user didn't actually send anything
-            req.bytes_transferred = 0
-        resp.last_modified = float(req.headers['X-Timestamp'])
-        return resp
+
+    def PUT_segmented_object(self, req, data_source, partition, nodes,
+                             container_partition, containers):
+        """
+        Performs a PUT for a segmented object (one with a content-length >
+        self.app.segment_size).
+
+        :param req: The webob.Request to process.
+        :param data_source: An iterator providing the data to store.
+        :param partition: The object ring partition the object falls on.
+        :param nodes: The object ring nodes the object falls on.
+        :param container_partition: The container ring partition the container
+                                    for the object falls on.
+        :param containers: The container ring nodes the container for the
+                           object falls on.
+        :returns: webob.Response object.
+        """
+        req.bytes_transferred = 0
+        leftover_chunk = [None]
+        etag = md5()
+
+        def segment_iter():
+            amount_given = 0
+            while amount_given < self.app.segment_size:
+                if leftover_chunk[0]:
+                    chunk = leftover_chunk[0]
+                    leftover_chunk[0] = None
+                else:
+                    with ChunkReadTimeout(self.app.client_timeout):
+                        chunk = data_source.next()
+                        req.bytes_transferred += len(chunk)
+                        etag.update(chunk)
+                if amount_given + len(chunk) > self.app.segment_size:
+                    yield chunk[:self.app.segment_size - amount_given]
+                    leftover_chunk[0] = \
+                        chunk[self.app.segment_size - amount_given:]
+                    amount_given = self.app.segment_size
+                else:
+                    yield chunk
+                    amount_given += len(chunk)
+
+        def segment_iter_iter():
+            while True:
+                if not leftover_chunk[0]:
+                    with ChunkReadTimeout(self.app.client_timeout):
+                        leftover_chunk[0] = data_source.next()
+                        req.bytes_transferred += len(leftover_chunk[0])
+                        etag.update(leftover_chunk[0])
+                yield segment_iter()
+
+        segment_number = 0
+        chunked = req.headers.get('transfer-encoding') == 'chunked'
+        if not chunked:
+            amount_left = req.content_length
+        headers = {'X-Timestamp': req.headers['X-Timestamp'],
+                   'Content-Type': req.headers['content-type'],
+                   'X-Object-Type': 'segment'}
+        for segment_source in segment_iter_iter():
+            if chunked:
+                headers['Transfer-Encoding'] = 'chunked'
+                if segment_number == 0:
+                    headers['X-Object-Segment-If-Length'] = \
+                        self.app.segment_size
+            elif amount_left > self.app.segment_size:
+                headers['Content-Length'] = self.app.segment_size
+            else:
+                headers['Content-Length'] = amount_left
+            headers['X-Object-Segment'] = segment_number
+            segment_req = Request.blank(req.path_info,
+                environ={'REQUEST_METHOD': 'PUT'}, headers=headers)
+            if 'X-Object-Segment-If-Length' in headers:
+                del headers['X-Object-Segment-If-Length']
+            if segment_number:
+                ring_object_name = '%s/%s/%s' % (self.object_name,
+                    req.headers['x-timestamp'], segment_number)
+            else:
+                ring_object_name = self.object_name
+            segment_partition, segment_nodes = self.app.object_ring.get_nodes(
+                self.account_name, self.container_name, ring_object_name)
+            resp = self.PUT_whole_object(segment_req, segment_source,
+                                         segment_partition, segment_nodes)
+            if resp.status_int // 100 == 4:
+                return resp
+            elif resp.status_int // 100 != 2:
+                return HTTPServiceUnavailable(request=req,
+                    body='Unable to complete very large file operation.')
+            if segment_number == 0 and \
+                    req.bytes_transferred < self.app.segment_size:
+                return HTTPCreated(request=req, etag=etag.hexdigest())
+            if not chunked:
+                amount_left -= self.app.segment_size
+            segment_number += 1
+        etag = etag.hexdigest()
+        if 'etag' in req.headers and req.headers['etag'].lower() != etag:
+            return HTTPUnprocessableEntity(request=req)
+        manifest = {'x-timestamp': req.headers['x-timestamp'],
+                    'content-length': req.bytes_transferred,
+                    'content-type': req.headers['content-type'],
+                    'x-segment-size': self.app.segment_size,
+                    'etag': etag}
+        if 'content-encoding' in req.headers:
+            manifest['content-encoding'] = req.headers['content-encoding']
+        manifest = pickle.dumps(manifest, protocol=PICKLE_PROTOCOL)
+        headers = {'X-Timestamp': req.headers['X-Timestamp'],
+                   'Content-Type': req.headers['content-type'],
+                   'Content-Length': len(manifest),
+                   'X-Object-Type': 'manifest',
+                   'X-Object-Length': req.bytes_transferred}
+        headers.update(i for i in req.headers.iteritems()
+            if i[0].lower().startswith('x-object-meta-') and len(i[0]) > 14)
+        manifest_req = Request.blank(req.path_info,
+            environ={'REQUEST_METHOD': 'PUT'}, body=manifest, headers=headers)
+        manifest_source = iter(lambda:
+            manifest_req.body_file.read(self.app.client_chunk_size), '')
+        resp = self.PUT_whole_object(manifest_req, manifest_source, partition,
+            nodes, container_partition=container_partition,
+            containers=containers)
+        if resp.status_int // 100 != 2:
+            return HTTPServiceUnavailable(request=req,
+                body='Unable to complete very large file operation.')
+        return HTTPCreated(request=req, etag=etag)
 
     @public
     @delay_denial
@@ -1233,6 +1596,8 @@
         if conf is None:
             conf = {}
         swift_dir = conf.get('swift_dir', '/etc/swift')
+        self.max_object_size = int(conf.get('max_object_size', 107374182400))
+        self.segment_size = int(conf.get('segment_size', 2147483647))
         self.node_timeout = int(conf.get('node_timeout', 10))
         self.conn_timeout = float(conf.get('conn_timeout', 0.5))
         self.client_timeout = int(conf.get('client_timeout', 60))

=== modified file 'test/functional/sample.conf'
--- test/functional/sample.conf	2010-09-09 17:24:25 +0000
+++ test/functional/sample.conf	2010-11-08 18:51:48 +0000
@@ -1,8 +1,15 @@
-# sample config
 auth_host = 127.0.0.1
 auth_port = 11000
 auth_ssl = no
 
+# The maximum object size the cluster allows (set in the proxy server's conf)
+max_object_size = 107374182400
+
+# The file segment size for the cluster (set in the proxy server's conf)
+# Set to 0 for no segment size testing (recommended if the segment size is
+# quite large and you don't want to spend the time testing it)
+segment_size = 0
+
 # Primary functional test account (needs admin access to the account)
 account = test
 username = tester

=== modified file 'test/functional/tests.py'
--- test/functional/tests.py	2010-10-29 20:30:34 +0000
+++ test/functional/tests.py	2010-11-08 18:51:48 +0000
@@ -1092,7 +1092,7 @@
             self.assert_(file.read(hdrs={'Range': r}) == data[0:1000])
 
     def testFileSizeLimit(self):
-        limit = 5*2**30 + 2
+        limit = int(config.get('max_object_size', 107374182400))
         tsecs = 3
 
         for i in (limit-100, limit-10, limit-1, limit, limit+1, limit+10,

=== modified file 'test/probe/common.py'
--- test/probe/common.py	2010-09-12 00:03:09 +0000
+++ test/probe/common.py	2010-11-08 18:51:48 +0000
@@ -88,7 +88,7 @@
     for p in ps:
         p.wait()
     ps = []
-    for job in ('container-updater', 'object-updater'):
+    for job in ('container-updater', 'object-janitor'):
         for n in xrange(1, 5):
             ps.append(Popen(['swift-%s' % job,
                              '/etc/swift/%s-server/%d.conf' %

=== modified file 'test/probe/test_object_async_update.py'
--- test/probe/test_object_async_update.py	2010-09-06 04:06:16 +0000
+++ test/probe/test_object_async_update.py	2010-11-08 18:51:48 +0000
@@ -55,7 +55,7 @@
                             self.account, container)[1])
         ps = []
         for n in xrange(1, 5):
-            ps.append(Popen(['swift-object-updater',
+            ps.append(Popen(['swift-object-janitor',
                              '/etc/swift/object-server/%d.conf' % n, 'once']))
         for p in ps:
             p.wait()

=== modified file 'test/unit/common/test_constraints.py'
--- test/unit/common/test_constraints.py	2010-08-16 22:30:27 +0000
+++ test/unit/common/test_constraints.py	2010-11-08 18:51:48 +0000
@@ -90,22 +90,20 @@
             headers=headers), 'object'), HTTPBadRequest))
 
     def test_check_object_creation_content_length(self):
-        headers = {'Content-Length': str(constraints.MAX_FILE_SIZE),
-                   'Content-Type': 'text/plain'}
+        headers = {'Content-Length': '1024', 'Content-Type': 'text/plain'}
         self.assertEquals(constraints.check_object_creation(Request.blank('/',
-            headers=headers), 'object_name'), None)
-        headers = {'Content-Length': str(constraints.MAX_FILE_SIZE + 1),
-                   'Content-Type': 'text/plain'}
+            headers=headers), 'object_name', 1024), None)
+        headers = {'Content-Length': '1025', 'Content-Type': 'text/plain'}
         self.assert_(isinstance(constraints.check_object_creation(
-            Request.blank('/', headers=headers), 'object_name'),
+            Request.blank('/', headers=headers), 'object_name', 1024),
             HTTPRequestEntityTooLarge))
         headers = {'Transfer-Encoding': 'chunked',
                    'Content-Type': 'text/plain'}
         self.assertEquals(constraints.check_object_creation(Request.blank('/',
-            headers=headers), 'object_name'), None)
+            headers=headers), 'object_name', 1024), None)
         headers = {'Content-Type': 'text/plain'}
         self.assert_(isinstance(constraints.check_object_creation(
-            Request.blank('/', headers=headers), 'object_name'),
+            Request.blank('/', headers=headers), 'object_name', 1024),
             HTTPLengthRequired))
 
     def test_check_object_creation_name_length(self):

=== added file 'test/unit/obj/test_diskfile.py'
--- test/unit/obj/test_diskfile.py	1970-01-01 00:00:00 +0000
+++ test/unit/obj/test_diskfile.py	2010-11-08 18:51:48 +0000
@@ -0,0 +1,203 @@
+# Copyright (c) 2010 OpenStack, LLC.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+# implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import cPickle as pickle
+import os
+import sys
+import unittest
+from nose import SkipTest
+from shutil import rmtree
+from StringIO import StringIO
+from time import time
+
+from xattr import setxattr
+
+from swift.common.constraints import PICKLE_PROTOCOL
+from swift.common.utils import mkdirs, normalize_timestamp
+from swift.obj.diskfile import DiskFile, hash_suffix, JANITORDIR, \
+                               METADATA_KEY
+
+
+class TestDiskFile(unittest.TestCase):
+    """ Test swift.obj.diskfile """
+
+    def setUp(self):
+        """ Set up for testing swift.obj.diskfile """
+        self.path_to_test_xfs = os.environ.get('PATH_TO_TEST_XFS')
+        if not self.path_to_test_xfs or \
+                not os.path.exists(self.path_to_test_xfs):
+            print >> sys.stderr, 'WARNING: PATH_TO_TEST_XFS not set or not ' \
+                'pointing to a valid directory.\n' \
+                'Please set PATH_TO_TEST_XFS to a directory on an XFS file ' \
+                'system for testing.'
+            self.testdir = '/tmp/SWIFTUNITTEST'
+        else:
+            self.testdir = os.path.join(self.path_to_test_xfs,
+                                        'tmp_test_obj_diskfile')
+        mkdirs(self.testdir)
+        rmtree(self.testdir)
+        mkdirs(os.path.join(self.testdir, 'sda1'))
+        mkdirs(os.path.join(self.testdir, 'sda1', 'tmp'))
+
+    def tearDown(self):
+        """ Tear down for testing swift.obj.diskfile """
+        rmtree(self.testdir)
+
+    def test_disk_file_app_iter_corners(self):
+        if not self.path_to_test_xfs:
+            raise SkipTest
+        df = DiskFile(self.testdir, 'sda1', '0', 'a', 'c', 'o')
+        mkdirs(df.datadir)
+        f = open(os.path.join(df.datadir,
+                              normalize_timestamp(time()) + '.data'), 'wb')
+        f.write('1234567890')
+        setxattr(f.fileno(), METADATA_KEY, pickle.dumps({}, PICKLE_PROTOCOL))
+        f.close()
+        df = DiskFile(self.testdir, 'sda1', '0', 'a', 'c', 'o',
+                      keep_data_fp=True)
+        it = df.app_iter_range(0, None)
+        sio = StringIO()
+        for chunk in it:
+            sio.write(chunk)
+        self.assertEquals(sio.getvalue(), '1234567890')
+
+        df = DiskFile(self.testdir, 'sda1', '0', 'a', 'c', 'o',
+                      keep_data_fp=True)
+        it = df.app_iter_range(5, None)
+        sio = StringIO()
+        for chunk in it:
+            sio.write(chunk)
+        self.assertEquals(sio.getvalue(), '67890')
+
+    def test_disk_file_mkstemp_creates_dir(self):
+        tmpdir = os.path.join(self.testdir, 'sda1', 'tmp')
+        os.rmdir(tmpdir)
+        with DiskFile(self.testdir, 'sda1', '0', 'a', 'c', 'o').mkstemp():
+            self.assert_(os.path.exists(tmpdir))
+
+    def test_hash_suffix_creates_janitor_jobs(self):
+        if not self.path_to_test_xfs:
+            raise SkipTest
+        # Ensure the janitor job we expect to create is not there right now.
+        df = DiskFile(self.testdir, 'sda1', '0', 'Segment-Cleanup',
+                      normalize_timestamp(1), 'a/c/o', datadir=JANITORDIR)
+        self.assert_(df.is_deleted())
+        # Set up manifest file to be tombstoned
+        df = DiskFile(self.testdir, 'sda1', '0', 'a', 'c', 'o')
+        with df.mkstemp() as (fd, tmppath):
+            os.write(fd, pickle.dumps({'x-timestamp': normalize_timestamp(1),
+                'content-length': 1234, 'content-type': 'text/plain',
+                'x-segment-size': 123,
+                'etag': 'd41d8cd98f00b204e9800998ecf8427e'},
+                protocol=PICKLE_PROTOCOL))
+            df.put(fd, tmppath, {'X-Timestamp': normalize_timestamp(2),
+                                 'X-Object-Type': 'manifest'})
+        # Make tombstone DiskFile didn't create, such as one rsynced over.
+        open(os.path.join(df.datadir, normalize_timestamp(3) + '.ts'), 'wb')
+        # Finally, we call what we want to test.
+        hash_suffix(os.path.dirname(df.datadir), 604800)
+        # Ensure the janitor job got created.
+        df = DiskFile(self.testdir, 'sda1', '0', 'Segment-Cleanup',
+                      normalize_timestamp(1), 'a/c/o', datadir=JANITORDIR)
+        self.assert_(not df.is_deleted())
+
+    def test_segment_info_overrides_datadir(self):
+        if not self.path_to_test_xfs:
+            raise SkipTest
+        df = DiskFile(self.testdir, 'sda1', '0', 'a', 'c', 'o')
+        datadir1 = df.datadir.split('/')[6]
+        df = DiskFile(self.testdir, 'sda1', '1', 'a', 'c', 'o')
+        datadir2 = df.datadir.split('/')[6]
+        df = DiskFile(self.testdir, 'sda1', '0', 'a', 'c', 'o', segment=0)
+        datadir3 = df.datadir.split('/')[6]
+        self.assertEquals(datadir1, datadir2)
+        self.assertNotEquals(datadir1, datadir3)
+
+    def test_no_longer_segment(self):
+        if not self.path_to_test_xfs:
+            raise SkipTest
+        # Normal case
+        df = DiskFile(self.testdir, 'sda1', '0', 'a', 'c', 'o')
+        datadir1 = df.datadir
+        with df.mkstemp() as (fd, tmppath):
+            df.put(fd, tmppath, {'X-Timestamp': normalize_timestamp(1)})
+        self.assert_(os.path.exists(datadir1))
+        # Normal case with no_longer_segment (doesn't make sense, but shouldn't
+        # blow up)
+        df = DiskFile(self.testdir, 'sda1', '1', 'a', 'c', 'o')
+        datadir1 = df.datadir
+        with df.mkstemp() as (fd, tmppath):
+            df.put(fd, tmppath, {'X-Timestamp': normalize_timestamp(1)},
+                   no_longer_segment=True)
+        self.assert_(os.path.exists(datadir1))
+        # Segment case
+        df = DiskFile(self.testdir, 'sda1', '2', 'a', 'c', 'o', segment=0)
+        datadir1 = df.datadir
+        with df.mkstemp() as (fd, tmppath):
+            df.put(fd, tmppath, {'X-Timestamp': normalize_timestamp(1)})
+        self.assert_(os.path.exists(datadir1))
+        # Segment case with no_longer_segment
+        df = DiskFile(self.testdir, 'sda1', '3', 'a', 'c', 'o')
+        normal_datadir1 = df.datadir
+        self.assert_(not os.path.exists(normal_datadir1))
+        df = DiskFile(self.testdir, 'sda1', '3', 'a', 'c', 'o', segment=0)
+        datadir1 = df.datadir
+        with df.mkstemp() as (fd, tmppath):
+            df.put(fd, tmppath, {'X-Timestamp': normalize_timestamp(1)},
+                   no_longer_segment=True)
+        self.assert_(not os.path.exists(datadir1))
+        self.assert_(os.path.exists(normal_datadir1))
+
+    def test_tombstone(self):
+        if not self.path_to_test_xfs:
+            raise SkipTest
+        df = DiskFile(self.testdir, 'sda1', '0', 'a', 'c', 'o')
+        with df.mkstemp() as (fd, tmppath):
+            df.put(fd, tmppath, {'X-Timestamp': normalize_timestamp(1)})
+        df = DiskFile(self.testdir, 'sda1', '0', 'a', 'c', 'o')
+        self.assert_(not df.is_deleted())
+        df = DiskFile(self.testdir, 'sda1', '0', 'a', 'c', 'o')
+        df.tombstone(normalize_timestamp(2))
+        df = DiskFile(self.testdir, 'sda1', '0', 'a', 'c', 'o')
+        self.assert_(df.is_deleted())
+
+    def test_store_janitor_container_update(self):
+        if not self.path_to_test_xfs:
+            raise SkipTest
+        df = DiskFile(self.testdir, 'sda1', '0', 'a', 'c', 'o',
+                      datadir=JANITORDIR)
+        df.store_janitor_container_update('PUT', 'a', 'c', 'o',
+            {'X-Timestamp': normalize_timestamp(1)}, [])
+        df = DiskFile(self.testdir, 'sda1', '0', 'a', 'c', 'o',
+                      datadir=JANITORDIR)
+        self.assertEquals(pickle.load(open(df.data_file, 'rb')),
+            {'op': 'PUT', 'account': 'a', 'container': 'c', 'obj': 'o',
+             'headers': {'X-Timestamp': '0000000001.00000'}, 'successes': []})
+
+    def test_store_janitor_segment_cleanup(self):
+        if not self.path_to_test_xfs:
+            raise SkipTest
+        df = DiskFile(self.testdir, 'sda1', '0', 'Segment-Cleanup',
+                normalize_timestamp(1), 'a/c/o', datadir=JANITORDIR)
+        df.store_janitor_segment_cleanup('a', 'c', 'o', 123, 45)
+        df = DiskFile(self.testdir, 'sda1', '0', 'Segment-Cleanup',
+                normalize_timestamp(1), 'a/c/o', datadir=JANITORDIR)
+        self.assertEquals(pickle.load(open(df.data_file, 'rb')),
+            {'account': 'a', 'container': 'c', 'obj': 'o',
+             'segment_count': 123, 'segment_last_deleted': 45})
+
+
+if __name__ == '__main__':
+    unittest.main()

=== added file 'test/unit/obj/test_janitor.py'
--- test/unit/obj/test_janitor.py	1970-01-01 00:00:00 +0000
+++ test/unit/obj/test_janitor.py	2010-11-08 18:51:48 +0000
@@ -0,0 +1,433 @@
+# Copyright (c) 2010 OpenStack, LLC.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+# implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import cPickle as pickle
+import os
+import sys
+import traceback
+import unittest
+from gzip import GzipFile
+from nose import SkipTest
+from shutil import rmtree
+from time import time
+
+from eventlet import spawn, TimeoutError, listen
+from eventlet.timeout import Timeout
+
+from swift.common.constraints import PICKLE_PROTOCOL
+from swift.common.ring import Ring, RingData
+from swift.common import utils
+from swift.common.utils import hash_path, normalize_timestamp, mkdirs
+from swift.obj import janitor as object_janitor
+from swift.obj.diskfile import DiskFile, JANITORDIR
+
+
+class TestObjectJanitor(unittest.TestCase):
+
+    def setUp(self):
+        utils.HASH_PATH_SUFFIX = 'endcap'
+        self.path_to_test_xfs = os.environ.get('PATH_TO_TEST_XFS')
+        if not self.path_to_test_xfs or \
+                not os.path.exists(self.path_to_test_xfs):
+            print >> sys.stderr, 'WARNING: PATH_TO_TEST_XFS not set or not ' \
+                'pointing to a valid directory.\n' \
+                'Please set PATH_TO_TEST_XFS to a directory on an XFS file ' \
+                'system for testing.'
+            self.testdir = '/tmp/SWIFTUNITTEST'
+        else:
+            self.testdir = os.path.join(self.path_to_test_xfs,
+                            'tmp_test_object_server_ObjectController')
+        rmtree(self.testdir, ignore_errors=1)
+        os.mkdir(self.testdir)
+        pickle.dump(RingData([[0, 1, 0, 1], [1, 0, 1, 0]],
+            [{'id': 0, 'ip': '127.0.0.1', 'port': 1, 'device': 'sda1',
+              'zone': 0},
+             {'id': 1, 'ip': '127.0.0.1', 'port': 1, 'device': 'sda1',
+              'zone': 2}], 30),
+            GzipFile(os.path.join(self.testdir, 'object.ring.gz'), 'wb'))
+        self.object_ring = Ring(os.path.join(self.testdir, 'object.ring.gz'))
+        pickle.dump(RingData([[0, 1, 0, 1], [1, 0, 1, 0]],
+            [{'id': 0, 'ip': '127.0.0.1', 'port': 1, 'device': 'sda1',
+              'zone': 0},
+             {'id': 1, 'ip': '127.0.0.1', 'port': 1, 'device': 'sda1',
+              'zone': 2}], 30),
+            GzipFile(os.path.join(self.testdir, 'container.ring.gz'), 'wb'))
+        self.container_ring = \
+            Ring(os.path.join(self.testdir, 'container.ring.gz'))
+        self.devices_dir = os.path.join(self.testdir, 'devices')
+        os.mkdir(self.devices_dir)
+        self.sda1 = os.path.join(self.devices_dir, 'sda1')
+        os.mkdir(self.sda1)
+        os.mkdir(os.path.join(self.sda1, 'tmp'))
+
+    def tearDown(self):
+        rmtree(self.testdir, ignore_errors=1)
+
+    def test_creation(self):
+        if not self.path_to_test_xfs:
+            raise SkipTest
+        janitor = object_janitor.ObjectJanitor({
+            'devices': self.devices_dir,
+            'mount_check': 'false',
+            'swift_dir': self.testdir,
+            'interval': '1',
+            'concurrency': '2',
+            'node_timeout': '5',
+            })
+        self.assert_(hasattr(janitor, 'logger'))
+        self.assert_(janitor.logger is not None)
+        self.assertEquals(janitor.devices, self.devices_dir)
+        self.assertEquals(janitor.interval, 1)
+        self.assertEquals(janitor.concurrency, 2)
+        self.assertEquals(janitor.node_timeout, 5)
+        self.assert_(janitor.get_container_ring() is not None)
+
+    def test_run_once_container_update(self):
+        if not self.path_to_test_xfs:
+            raise SkipTest
+        janitor = object_janitor.ObjectJanitor({
+            'devices': self.devices_dir,
+            'mount_check': 'false',
+            'swift_dir': self.testdir,
+            'interval': '1',
+            'concurrency': '1',
+            'node_timeout': '15',
+            })
+        janitor.run_once()
+        janitor_dir = os.path.join(self.sda1, JANITORDIR)
+        os.mkdir(janitor_dir)
+        janitor.run_once()
+        self.assert_(os.path.exists(janitor_dir))
+
+        disk_file = DiskFile(self.devices_dir, 'sda1',
+            str(self.container_ring.get_nodes('a', 'c', 'o')[0]), 'a', 'c',
+            'o', datadir=JANITORDIR)
+        ts = normalize_timestamp(1)
+        with disk_file.mkstemp() as (fd, tmppath):
+            os.write(fd, pickle.dumps({'op': 'PUT',
+                'account': 'a', 'container': 'c', 'obj': 'o',
+                'headers': {'X-Container-Timestamp': ts}},
+                PICKLE_PROTOCOL))
+            disk_file.put(fd, tmppath,
+                          {'X-Op': 'Container-Update', 'X-Timestamp': ts})
+        janitor.run_once()
+        self.assert_(os.path.exists(os.path.join(disk_file.datadir,
+                                                 ts + '.data')))
+
+        bindsock = listen(('127.0.0.1', 0))
+
+        def accepter(sock, return_code):
+            try:
+                with Timeout(3):
+                    inc = sock.makefile('rb')
+                    out = sock.makefile('wb')
+                    out.write('HTTP/1.1 %d OK\r\nContent-Length: 0\r\n\r\n' %
+                              return_code)
+                    out.flush()
+                    self.assertEquals(inc.readline(),
+                                      'PUT /sda1/0/a/c/o HTTP/1.1\r\n')
+                    headers = {}
+                    line = inc.readline()
+                    while line and line != '\r\n':
+                        headers[line.split(':')[0].lower()] = \
+                            line.split(':')[1].strip()
+                        line = inc.readline()
+                    self.assert_('x-container-timestamp' in headers)
+            except BaseException, err:
+                return err
+            return None
+
+        def accept(return_codes):
+            codes = iter(return_codes)
+            try:
+                events = []
+                for x in xrange(len(return_codes)):
+                    with Timeout(3):
+                        sock, addr = bindsock.accept()
+                        events.append(
+                            spawn(accepter, sock, codes.next()))
+                for event in events:
+                    err = event.wait()
+                    if err:
+                        raise err
+            except BaseException, err:
+                return err
+            return None
+
+        event = spawn(accept, [201, 500])
+        for dev in janitor.get_container_ring().devs:
+            if dev is not None:
+                dev['port'] = bindsock.getsockname()[1]
+        janitor.run_once()
+        err = event.wait()
+        if err:
+            raise err
+        disk_file = DiskFile(self.devices_dir, 'sda1',
+            str(self.container_ring.get_nodes('a', 'c', 'o')[0]), 'a', 'c',
+            'o', datadir=JANITORDIR)
+        self.assert_(not disk_file.is_deleted())
+        event = spawn(accept, [201])
+        janitor.run_once()
+        err = event.wait()
+        if err:
+            raise err
+        disk_file = DiskFile(self.devices_dir, 'sda1',
+            str(self.container_ring.get_nodes('a', 'c', 'o')[0]), 'a', 'c',
+            'o', datadir=JANITORDIR)
+        self.assert_(disk_file.is_deleted())
+
+    def _segment_cleanup_in_progress_helper(self, statuses, expect_success):
+        if not self.path_to_test_xfs:
+            raise SkipTest
+        janitor = object_janitor.ObjectJanitor({'devices': self.devices_dir,
+            'mount_check': 'false', 'swift_dir': self.testdir, 'interval': '1',
+            'concurrency': '1', 'node_timeout': '15'})
+
+        # Quick test of connection refusals
+        df = DiskFile(self.devices_dir, 'sda1',
+            str(self.object_ring.get_nodes('a', 'c', 'o')[0]),
+            'Segment-Cleanup', normalize_timestamp(1), 'a/c/o',
+            datadir=JANITORDIR)
+        df.store_janitor_segment_cleanup('a', 'c', 'o', 2, 0)
+        janitor.run_once()
+        df = DiskFile(self.devices_dir, 'sda1',
+            str(self.object_ring.get_nodes('a', 'c', 'o')[0]),
+            'Segment-Cleanup', normalize_timestamp(1), 'a/c/o',
+            datadir=JANITORDIR)
+        self.assert_(not df.is_deleted())
+
+        bindsock = listen(('127.0.0.1', 0))
+        janitor.port = bindsock.getsockname()[1]
+
+        def accepter(sock, return_code):
+            try:
+                with Timeout(3):
+                    inc = sock.makefile('rb')
+                    out = sock.makefile('wb')
+                    out.write('HTTP/1.1 %d OK\r\nContent-Length: 0\r\n\r\n' %
+                              return_code)
+                    out.flush()
+                    self.assertEquals(inc.readline(),
+                                      'DELETE /sda1/2/a/c/o HTTP/1.1\r\n')
+                    headers = {}
+                    line = inc.readline()
+                    while line and line != '\r\n':
+                        headers[line.split(':')[0].lower()] = \
+                            line.split(':')[1].strip()
+                        line = inc.readline()
+                    self.assert_('x-object-segment-timestamp' in headers)
+                    self.assertEquals(headers.get('x-object-segment'), '1')
+            except BaseException, err:
+                return err
+            return None
+
+        def accept(return_codes):
+            codes = iter(return_codes)
+            try:
+                events = []
+                for x in xrange(len(return_codes)):
+                    with Timeout(3):
+                        sock, addr = bindsock.accept()
+                        events.append(
+                            spawn(accepter, sock, codes.next()))
+                for event in events:
+                    err = event.wait()
+                    if err:
+                        raise err
+            except BaseException, err:
+                return err
+            return None
+
+        event = spawn(accept, statuses)
+        for dev in janitor.get_object_ring().devs:
+            if dev is not None:
+                dev['port'] = bindsock.getsockname()[1]
+        janitor.run_once()
+        err = event.wait()
+        if err:
+            raise err
+        df = DiskFile(self.devices_dir, 'sda1',
+            str(self.object_ring.get_nodes('a', 'c', 'o')[0]),
+            'Segment-Cleanup', normalize_timestamp(1), 'a/c/o',
+            datadir=JANITORDIR)
+        self.assertEquals(df.is_deleted(), expect_success)
+
+    def test_segment_cleanup_in_progress_happy_path(self):
+        self._segment_cleanup_in_progress_helper([204, 204], True)
+
+    def test_segment_cleanup_in_progress_one_failure(self):
+        self._segment_cleanup_in_progress_helper([204, 500], True)
+
+    def test_segment_cleanup_in_progress_all_failures(self):
+        self._segment_cleanup_in_progress_helper([500, 500], False)
+
+    def test_segment_cleanup_in_progress_all_not_found(self):
+        self._segment_cleanup_in_progress_helper([404, 404], True)
+
+    def test_segment_cleanup_in_progress_one_not_found_one_success(self):
+        self._segment_cleanup_in_progress_helper([404, 204], True)
+
+    def test_segment_cleanup_in_progress_one_not_found_one_failure(self):
+        self._segment_cleanup_in_progress_helper([404, 500], False)
+
+    def _segment_cleanup_fresh_start_helper(self, cleanup_timestamp,
+            existing_timestamp, statuses, expect_success):
+        if not self.path_to_test_xfs:
+            raise SkipTest
+        janitor = object_janitor.ObjectJanitor({'devices': self.devices_dir,
+            'mount_check': 'false', 'swift_dir': self.testdir, 'interval': '1',
+            'concurrency': '1', 'node_timeout': '15', 'segments_per_pass': 2})
+
+        # Quick test of connection refusals
+        df = DiskFile(self.devices_dir, 'sda1',
+            str(self.object_ring.get_nodes('a', 'c', 'o')[0]),
+            'Segment-Cleanup', normalize_timestamp(cleanup_timestamp), 'a/c/o',
+            datadir=JANITORDIR)
+        df.store_janitor_segment_cleanup('a', 'c', 'o', None, None)
+        janitor.run_once()
+        df = DiskFile(self.devices_dir, 'sda1',
+            str(self.object_ring.get_nodes('a', 'c', 'o')[0]),
+            'Segment-Cleanup', normalize_timestamp(cleanup_timestamp), 'a/c/o',
+            datadir=JANITORDIR)
+        self.assert_(not df.is_deleted())
+
+        bindsock = listen(('127.0.0.1', 0))
+        janitor.port = bindsock.getsockname()[1]
+
+        def accepter(sock, return_code):
+            try:
+                with Timeout(3):
+                    inc = sock.makefile('rb')
+                    out = sock.makefile('wb')
+                    request = inc.readline()
+                    if request.startswith('GET '):
+                        self.assert_(request, 'GET /sda1/0/a/c/o HTTP/1.1\r\n')
+                        if return_code == 200:
+                            pickl = pickle.dumps({'x-timestamp':
+                                normalize_timestamp(existing_timestamp)})
+                            out.write('HTTP/1.1 %d OK\r\nContent-Length: '
+                                '%s\r\nX-Object-Type: manifest\r\n\r\n' %
+                                (return_code, len(pickl)))
+                            out.write(pickl)
+                            out.flush()
+                        else:
+                            out.write('HTTP/1.1 %d OK\r\nContent-Length: '
+                                '0\r\n\r\n' % return_code)
+                            out.flush()
+                        return None
+                    if request.startswith('HEAD '):
+                        self.assert_(request,
+                                     'HEAD /sda1/2/a/c/o HTTP/1.1\r\n')
+                        out.write('HTTP/1.1 %d OK\r\nContent-Length: '
+                            '0\r\n\r\n' % return_code)
+                        out.flush()
+                        return None
+                    else:
+                        self.assert_(request,
+                                     'DELETE /sda1/2/a/c/o HTTP/1.1\r\n')
+                        out.write(
+                            'HTTP/1.1 %d OK\r\nContent-Length: 0\r\n\r\n' %
+                            return_code)
+                        out.flush()
+                    headers = {}
+                    line = inc.readline()
+                    while line and line != '\r\n':
+                        headers[line.split(':')[0].lower()] = \
+                            line.split(':')[1].strip()
+                        line = inc.readline()
+                    self.assert_('x-object-segment' in headers)
+                    self.assert_('x-object-segment-timestamp' in headers)
+            except BaseException, err:
+                traceback.print_exc()
+                return err
+            return None
+
+        def accept(return_codes):
+            codes = iter(return_codes)
+            try:
+                events = []
+                for x in xrange(len(return_codes)):
+                    with Timeout(3):
+                        sock, addr = bindsock.accept()
+                        events.append(
+                            spawn(accepter, sock, codes.next()))
+                for event in events:
+                    err = event.wait()
+                    if err:
+                        raise err
+            except BaseException, err:
+                return err
+            return None
+
+        event = spawn(accept, statuses)
+        for dev in janitor.get_object_ring().devs:
+            if dev is not None:
+                dev['port'] = bindsock.getsockname()[1]
+        janitor.run_once()
+        err = event.wait()
+        if err:
+            raise err
+        df = DiskFile(self.devices_dir, 'sda1',
+            str(self.object_ring.get_nodes('a', 'c', 'o')[0]),
+            'Segment-Cleanup', normalize_timestamp(cleanup_timestamp), 'a/c/o',
+            datadir=JANITORDIR)
+        self.assertEquals(df.is_deleted(), expect_success)
+
+    def test_segment_cleanup_fresh_start_happy_path(self):
+        self._segment_cleanup_fresh_start_helper(normalize_timestamp(1),
+            normalize_timestamp(1),
+            [404, 404,              # Check for manifest
+             200, 200, 404, 404,    # Check for segments (2 total)
+             204, 204, 204, 204],   # Delete segments
+            True)
+
+    def test_segment_cleanup_fresh_start_manifest_exists(self):
+        t = time()
+        self._segment_cleanup_fresh_start_helper(normalize_timestamp(t),
+            normalize_timestamp(t), [200, 200], True)
+
+    def test_segment_cleanup_fresh_start_old_manifest_exists(self):
+        self._segment_cleanup_fresh_start_helper(normalize_timestamp(2),
+            normalize_timestamp(1),
+            [200, 200,              # Check for manifest
+             200, 200, 404, 404,    # Check for segments (2 total)
+             204, 204, 204, 204],   # Delete segments
+            True)
+
+    def test_segment_cleanup_fresh_start_old_manifest_exists2(self):
+        self._segment_cleanup_fresh_start_helper(normalize_timestamp(time()),
+            normalize_timestamp(1), [200, 200], False)
+
+    def test_segment_cleanup_fresh_start_new_manifest_exists(self):
+        t = time()
+        self._segment_cleanup_fresh_start_helper(normalize_timestamp(t - 1),
+            normalize_timestamp(t),
+            [200,                   # Check for manifest
+             200, 200, 404, 404,    # Check for segments (2 total)
+             204, 204, 204, 204],   # Delete segments
+            True)
+
+    def test_segment_cleanup_fresh_start_stops_at_segments_per_pass(self):
+        self._segment_cleanup_fresh_start_helper(normalize_timestamp(1),
+            normalize_timestamp(1),
+            [404, 404,                  # Check for manifest
+             200, 200, 200, 404, 404,   # Check for segments (3 total)
+             204, 204, 204, 204],       # Delete segments (only 2 expected)
+            False)
+
+
+
+if __name__ == '__main__':
+    unittest.main()

=== modified file 'test/unit/obj/test_server.py'
--- test/unit/obj/test_server.py	2010-10-13 21:26:43 +0000
+++ test/unit/obj/test_server.py	2010-11-08 18:51:48 +0000
@@ -13,7 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-""" Tests for swift.object_server """
+""" Tests for swift.obj.server """
 
 import cPickle as pickle
 import os
@@ -22,23 +22,25 @@
 from nose import SkipTest
 from shutil import rmtree
 from StringIO import StringIO
-from time import gmtime, sleep, strftime, time
+from time import gmtime, strftime, time
 
 from eventlet import sleep, spawn, wsgi, listen
 from webob import Request
-from xattr import getxattr, setxattr
+from xattr import getxattr
 
 from test.unit import connect_tcp, readuntil2crlfs
+from swift.common.constraints import PICKLE_PROTOCOL
 from swift.obj import server as object_server
+from swift.obj.diskfile import DATADIR, DiskFile, JANITORDIR, METADATA_KEY
 from swift.common.utils import hash_path, mkdirs, normalize_timestamp, \
                                NullLogger, storage_directory
 
 
 class TestObjectController(unittest.TestCase):
-    """ Test swift.object_server.ObjectController """
+    """ Test swift.obj.server.ObjectController """
 
     def setUp(self):
-        """ Set up for testing swift.object_server.ObjectController """
+        """ Set up for testing swift.obj.server.ObjectController """
         self.path_to_test_xfs = os.environ.get('PATH_TO_TEST_XFS')
         if not self.path_to_test_xfs or \
                 not os.path.exists(self.path_to_test_xfs):
@@ -49,7 +51,7 @@
             self.testdir = '/tmp/SWIFTUNITTEST'
         else:
             self.testdir = os.path.join(self.path_to_test_xfs,
-                            'tmp_test_object_server_ObjectController')
+                                        'tmp_test_obj_server_ObjectController')
         mkdirs(self.testdir)
         rmtree(self.testdir)
         mkdirs(os.path.join(self.testdir, 'sda1'))
@@ -59,11 +61,11 @@
         self.object_controller.bytes_per_sync = 1
 
     def tearDown(self):
-        """ Tear down for testing swift.object_server.ObjectController """
+        """ Tear down for testing swift.obj.server.ObjectController """
         rmtree(self.testdir)
 
     def test_POST_update_meta(self):
-        """ Test swift.object_server.ObjectController.POST """
+        """ Test swift.obj.server.ObjectController.POST """
         if not self.path_to_test_xfs:
             raise SkipTest
         timestamp = normalize_timestamp(time())
@@ -221,12 +223,11 @@
         resp = self.object_controller.PUT(req)
         self.assertEquals(resp.status_int, 201)
         objfile = os.path.join(self.testdir, 'sda1',
-            storage_directory(object_server.DATADIR, 'p',
-                              hash_path('a', 'c', 'o')),
+            storage_directory(DATADIR, 'p', hash_path('a', 'c', 'o')),
             timestamp + '.data')
         self.assert_(os.path.isfile(objfile))
         self.assertEquals(open(objfile).read(), 'VERIFY')
-        self.assertEquals(pickle.loads(getxattr(objfile, object_server.METADATA_KEY)),
+        self.assertEquals(pickle.loads(getxattr(objfile, METADATA_KEY)),
                           {'X-Timestamp': timestamp,
                            'Content-Length': '6',
                            'ETag': '0b4c12d7e0a73840c1c4f148fda3b037',
@@ -253,12 +254,11 @@
         resp = self.object_controller.PUT(req)
         self.assertEquals(resp.status_int, 201)
         objfile = os.path.join(self.testdir, 'sda1',
-            storage_directory(object_server.DATADIR, 'p',
-                              hash_path('a', 'c', 'o')),
+            storage_directory(DATADIR, 'p', hash_path('a', 'c', 'o')),
             timestamp + '.data')
         self.assert_(os.path.isfile(objfile))
         self.assertEquals(open(objfile).read(), 'VERIFY TWO')
-        self.assertEquals(pickle.loads(getxattr(objfile, object_server.METADATA_KEY)),
+        self.assertEquals(pickle.loads(getxattr(objfile, METADATA_KEY)),
                           {'X-Timestamp': timestamp,
                            'Content-Length': '10',
                            'ETag': 'b381a4c5dab1eaa1eb9711fa647cd039',
@@ -299,12 +299,11 @@
         resp = self.object_controller.PUT(req)
         self.assertEquals(resp.status_int, 201)
         objfile = os.path.join(self.testdir, 'sda1',
-            storage_directory(object_server.DATADIR, 'p',
-                              hash_path('a', 'c', 'o')),
+            storage_directory(DATADIR, 'p', hash_path('a', 'c', 'o')),
             timestamp + '.data')
         self.assert_(os.path.isfile(objfile))
         self.assertEquals(open(objfile).read(), 'VERIFY THREE')
-        self.assertEquals(pickle.loads(getxattr(objfile, object_server.METADATA_KEY)),
+        self.assertEquals(pickle.loads(getxattr(objfile, METADATA_KEY)),
                           {'X-Timestamp': timestamp,
                            'Content-Length': '12',
                            'ETag': 'b114ab7b90d9ccac4bd5d99cc7ebb568',
@@ -375,7 +374,7 @@
             object_server.http_connect = old_http_connect
 
     def test_HEAD(self):
-        """ Test swift.object_server.ObjectController.HEAD """
+        """ Test swift.obj.server.ObjectController.HEAD """
         if not self.path_to_test_xfs:
             raise SkipTest
         req = Request.blank('/sda1/p/a/c')
@@ -410,8 +409,7 @@
         self.assertEquals(resp.headers['x-object-meta-two'], 'Two')
 
         objfile = os.path.join(self.testdir, 'sda1',
-            storage_directory(object_server.DATADIR, 'p',
-                              hash_path('a', 'c', 'o')),
+            storage_directory(DATADIR, 'p', hash_path('a', 'c', 'o')),
             timestamp + '.data')
         os.unlink(objfile)
         req = Request.blank('/sda1/p/a/c/o')
@@ -442,7 +440,7 @@
         self.assertEquals(resp.status_int, 404)
 
     def test_GET(self):
-        """ Test swift.object_server.ObjectController.GET """
+        """ Test swift.obj.server.ObjectController.GET """
         if not self.path_to_test_xfs:
             raise SkipTest
         req = Request.blank('/sda1/p/a/c')
@@ -500,8 +498,7 @@
         self.assertEquals(resp.headers['content-length'], '2')
 
         objfile = os.path.join(self.testdir, 'sda1',
-            storage_directory(object_server.DATADIR, 'p',
-                              hash_path('a', 'c', 'o')),
+            storage_directory(DATADIR, 'p', hash_path('a', 'c', 'o')),
             timestamp + '.data')
         os.unlink(objfile)
         req = Request.blank('/sda1/p/a/c/o')
@@ -712,7 +709,7 @@
         self.assertEquals(resp.status_int, 200)
 
     def test_DELETE(self):
-        """ Test swift.object_server.ObjectController.DELETE """
+        """ Test swift.obj.server.ObjectController.DELETE """
         if not self.path_to_test_xfs:
             raise SkipTest
         req = Request.blank('/sda1/p/a/c', environ={'REQUEST_METHOD': 'DELETE'})
@@ -751,8 +748,7 @@
         resp = self.object_controller.DELETE(req)
         self.assertEquals(resp.status_int, 204)
         objfile = os.path.join(self.testdir, 'sda1',
-            storage_directory(object_server.DATADIR, 'p',
-                              hash_path('a', 'c', 'o')),
+            storage_directory(DATADIR, 'p', hash_path('a', 'c', 'o')),
             timestamp + '.ts')
         self.assert_(os.path.isfile(objfile))
 
@@ -764,13 +760,12 @@
         resp = self.object_controller.DELETE(req)
         self.assertEquals(resp.status_int, 204)
         objfile = os.path.join(self.testdir, 'sda1',
-            storage_directory(object_server.DATADIR, 'p',
-                              hash_path('a', 'c', 'o')),
+            storage_directory(DATADIR, 'p', hash_path('a', 'c', 'o')),
             timestamp + '.ts')
         self.assert_(os.path.isfile(objfile))
 
     def test_call(self):
-        """ Test swift.object_server.ObjectController.__call__ """
+        """ Test swift.obj.server.ObjectController.__call__ """
         inbuf = StringIO()
         errbuf = StringIO()
         outbuf = StringIO()
@@ -886,39 +881,6 @@
         resp = self.object_controller.PUT(req)
         self.assertEquals(resp.status_int, 400)
 
-    def test_disk_file_app_iter_corners(self):
-        if not self.path_to_test_xfs:
-            raise SkipTest
-        df = object_server.DiskFile(self.testdir, 'sda1', '0', 'a', 'c', 'o')
-        mkdirs(df.datadir)
-        f = open(os.path.join(df.datadir,
-                              normalize_timestamp(time()) + '.data'), 'wb')
-        f.write('1234567890')
-        setxattr(f.fileno(), object_server.METADATA_KEY,
-                 pickle.dumps({}, object_server.PICKLE_PROTOCOL))
-        f.close()
-        df = object_server.DiskFile(self.testdir, 'sda1', '0', 'a', 'c', 'o',
-                                    keep_data_fp=True)
-        it = df.app_iter_range(0, None)
-        sio = StringIO()
-        for chunk in it:
-            sio.write(chunk)
-        self.assertEquals(sio.getvalue(), '1234567890')
-
-        df = object_server.DiskFile(self.testdir, 'sda1', '0', 'a', 'c', 'o',
-                                    keep_data_fp=True)
-        it = df.app_iter_range(5, None)
-        sio = StringIO()
-        for chunk in it:
-            sio.write(chunk)
-        self.assertEquals(sio.getvalue(), '67890')
-
-    def test_disk_file_mkstemp_creates_dir(self):
-        tmpdir = os.path.join(self.testdir, 'sda1', 'tmp')
-        os.rmdir(tmpdir)
-        with object_server.DiskFile(self.testdir, 'sda1', '0', 'a', 'c', 'o').mkstemp():
-            self.assert_(os.path.exists(tmpdir))
-
     def test_max_upload_time(self):
         if not self.path_to_test_xfs:
             raise SkipTest
@@ -1006,6 +968,161 @@
         self.assertEquals(resp.status_int, 200)
         self.assertEquals(resp.headers['content-encoding'], 'gzip')
 
+    def test_overwritten_manifest(self):
+        """
+        Ensures a janitor segment cleanup job is created for an overwritten
+        manifest.
+        """
+        if not self.path_to_test_xfs:
+            raise SkipTest
+        df = DiskFile(self.testdir, 'sda1', '0', 'Segment-Cleanup',
+                      normalize_timestamp(1), 'a/c/o', datadir=JANITORDIR)
+        self.assert_(df.is_deleted())
+
+        manifest = {'x-timestamp': normalize_timestamp(1),
+            'content-length': 100, 'content-type': 'text/plain',
+            'x-segment-size': 10, 'etag': 'd41d8cd98f00b204e9800998ecf8427e'}
+        manifest = pickle.dumps(manifest, protocol=PICKLE_PROTOCOL)
+        req = Request.blank('/sda1/0/a/c/o', environ={'REQUEST_METHOD': 'PUT'},
+            headers={'X-Timestamp': normalize_timestamp(1),
+            'Content-Length': str(len(manifest)), 'Content-Type': 'text/plain',
+            'X-Object-Type': 'manifest', 'X-Object-Length': '100'},
+            body=manifest)
+        resp = self.object_controller.PUT(req)
+        self.assertEquals(resp.status_int, 201)
+
+        req = Request.blank('/sda1/0/a/c/o', environ={'REQUEST_METHOD': 'PUT'},
+            headers={'X-Timestamp': normalize_timestamp(2),
+            'Content-Length': '1', 'Content-Type': 'text/plain'}, body=' ')
+        resp = self.object_controller.PUT(req)
+        self.assertEquals(resp.status_int, 201)
+
+        df = DiskFile(self.testdir, 'sda1', '0', 'Segment-Cleanup',
+                      normalize_timestamp(1), 'a/c/o', datadir=JANITORDIR)
+        self.assert_(not df.is_deleted())
+
+    def test_segmented_put(self):
+        if not self.path_to_test_xfs:
+            raise SkipTest
+
+        # Ensure the janitor cleanup job doesn't exist yet
+        df = DiskFile(self.testdir, 'sda1', '0', 'Segment-Cleanup',
+            normalize_timestamp(1), 'a/c/o', datadir=JANITORDIR)
+        self.assert_(df.is_deleted())
+
+        # Put the first segment
+        req = Request.blank('/sda1/0/a/c/o', environ={'REQUEST_METHOD': 'PUT'},
+            headers={'X-Timestamp': normalize_timestamp(1),
+            'Content-Length': '1', 'Content-Type': 'text/plain',
+            'X-Object-Type': 'segment', 'X-Object-Segment': '0',
+            'X-Object-Segment-If-Length': '1'}, body='1')
+        resp = self.object_controller.PUT(req)
+        self.assertEquals(resp.status_int, 201)
+
+        # Ensure the janitor cleanup job now exists
+        df = DiskFile(self.testdir, 'sda1', '0', 'Segment-Cleanup',
+            normalize_timestamp(1), 'a/c/o', datadir=JANITORDIR)
+        self.assert_(not df.is_deleted())
+
+        # Second segment would go to a different node
+
+        # Put the manifest
+        manifest = {'x-timestamp': normalize_timestamp(1),
+            'content-length': 2, 'content-type': 'text/plain',
+            'x-segment-size': 1, 'etag': 'c20ad4d76fe97759aa27a0c99bff6710'}
+        manifest = pickle.dumps(manifest, protocol=PICKLE_PROTOCOL)
+        req = Request.blank('/sda1/0/a/c/o', environ={'REQUEST_METHOD': 'PUT'},
+            headers={'X-Timestamp': normalize_timestamp(1),
+            'Content-Length': str(len(manifest)), 'Content-Type': 'text/plain',
+            'X-Object-Type': 'manifest', 'X-Object-Length': '2'},
+            body=manifest)
+        resp = self.object_controller.PUT(req)
+        self.assertEquals(resp.status_int, 201)
+
+        # The janitor cleanup job should still exist (only the janitor will
+        # verify the manifest is in place an remove the job).
+        df = DiskFile(self.testdir, 'sda1', '0', 'Segment-Cleanup',
+            normalize_timestamp(1), 'a/c/o', datadir=JANITORDIR)
+        self.assert_(not df.is_deleted())
+
+        req = Request.blank('/sda1/0/a/c/o', environ={'REQUEST_METHOD': 'GET'})
+        resp = self.object_controller.GET(req)
+        self.assertEquals(resp.status_int, 200)
+        self.assertEquals(resp.body, manifest)
+
+        req = Request.blank('/sda1/0/a/c/o', environ={'REQUEST_METHOD': 'GET'},
+            headers={'X-Object-Segment': '0',
+                     'X-Object-Segment-Timestamp': normalize_timestamp(1)})
+        resp = self.object_controller.GET(req)
+        self.assertEquals(resp.status_int, 200)
+        self.assertEquals(resp.body, '1')
+
+    def test_segmented_put_no_longer(self):
+        if not self.path_to_test_xfs:
+            raise SkipTest
+
+        # Ensure the janitor cleanup job doesn't exist to begin with
+        df = DiskFile(self.testdir, 'sda1', '0', 'Segment-Cleanup',
+            normalize_timestamp(1), 'a/c/o', datadir=JANITORDIR)
+        self.assert_(df.is_deleted())
+
+        # Put the first segment, that really ends up a whole object
+        req = Request.blank('/sda1/0/a/c/o', environ={'REQUEST_METHOD': 'PUT'},
+            headers={'X-Timestamp': normalize_timestamp(1),
+            'Content-Length': '1', 'Content-Type': 'text/plain',
+            'X-Object-Type': 'segment', 'X-Object-Segment': '0',
+            'X-Object-Segment-If-Length': '2'}, body='1')
+        resp = self.object_controller.PUT(req)
+        self.assertEquals(resp.status_int, 201)
+
+        # Ensure the janitor cleanup job doesn't exist since we put a whole
+        # file, not a segment (due to X-Object-Segment-If-Length).
+        df = DiskFile(self.testdir, 'sda1', '0', 'Segment-Cleanup',
+            normalize_timestamp(1), 'a/c/o', datadir=JANITORDIR)
+        self.assert_(df.is_deleted())
+
+        req = Request.blank('/sda1/0/a/c/o', environ={'REQUEST_METHOD': 'GET'})
+        resp = self.object_controller.GET(req)
+        self.assertEquals(resp.status_int, 200)
+        self.assertEquals(resp.body, '1')
+
+    def test_deleted_manifest(self):
+        """
+        Ensures a janitor segment cleanup job is created for a deleted
+        manifest.
+        """
+        if not self.path_to_test_xfs:
+            raise SkipTest
+        df = DiskFile(self.testdir, 'sda1', '0', 'Segment-Cleanup',
+                      normalize_timestamp(1), 'a/c/o', datadir=JANITORDIR)
+        self.assert_(df.is_deleted())
+
+        manifest = {'x-timestamp': normalize_timestamp(1),
+            'content-length': 100, 'content-type': 'text/plain',
+            'x-segment-size': 10, 'etag': 'd41d8cd98f00b204e9800998ecf8427e'}
+        manifest = pickle.dumps(manifest, protocol=PICKLE_PROTOCOL)
+        req = Request.blank('/sda1/0/a/c/o', environ={'REQUEST_METHOD': 'PUT'},
+            headers={'X-Timestamp': normalize_timestamp(1),
+            'Content-Length': str(len(manifest)), 'Content-Type': 'text/plain',
+            'X-Object-Type': 'manifest', 'X-Object-Length': '100'},
+            body=manifest)
+        resp = self.object_controller.PUT(req)
+        self.assertEquals(resp.status_int, 201)
+
+        req = Request.blank('/sda1/0/a/c/o',
+            environ={'REQUEST_METHOD': 'DELETE'},
+            headers={'X-Timestamp': normalize_timestamp(2),
+            'Content-Length': '0', 'Content-Type': 'text/plain'}, body='')
+        resp = self.object_controller.DELETE(req)
+        self.assertEquals(resp.status_int, 204)
+
+        df = DiskFile(self.testdir, 'sda1', '0', 'Segment-Cleanup',
+            normalize_timestamp(1), 'a/c/o', datadir=JANITORDIR,
+            keep_data_fp=True)
+        self.assert_(not df.is_deleted())
+        job = pickle.loads(''.join(iter(df)))
+        self.assertEquals(job['segment_last_deleted'], -1)
+
 
 if __name__ == '__main__':
     unittest.main()

=== modified file 'test/unit/obj/test_updater.py'
--- test/unit/obj/test_updater.py	2010-09-23 16:09:30 +0000
+++ test/unit/obj/test_updater.py	2010-11-08 18:51:48 +0000
@@ -24,7 +24,7 @@
 from eventlet import spawn, TimeoutError, listen
 from eventlet.timeout import Timeout
 
-from swift.obj import updater as object_updater, server as object_server
+from swift.obj import updater as object_updater
 from swift.common.ring import RingData
 from swift.common import utils
 from swift.common.utils import hash_path, normalize_timestamp, mkdirs
@@ -48,7 +48,7 @@
         os.mkdir(self.devices_dir)
         self.sda1 = os.path.join(self.devices_dir, 'sda1')
         os.mkdir(self.sda1)
-        os.mkdir(os.path.join(self.sda1,'tmp'))
+        os.mkdir(os.path.join(self.sda1, 'tmp'))
 
     def tearDown(self):
         rmtree(self.testdir, ignore_errors=1)
@@ -80,7 +80,7 @@
             'node_timeout': '15',
             })
         cu.run_once()
-        async_dir = os.path.join(self.sda1, object_server.ASYNCDIR)
+        async_dir = os.path.join(self.sda1, object_updater.ASYNCDIR)
         os.mkdir(async_dir)
         cu.run_once()
         self.assert_(os.path.exists(async_dir))
@@ -103,6 +103,7 @@
         self.assert_(os.path.exists(op_path))
 
         bindsock = listen(('127.0.0.1', 0))
+
         def accepter(sock, return_code):
             try:
                 with Timeout(3):
@@ -123,6 +124,7 @@
             except BaseException, err:
                 return err
             return None
+
         def accept(return_codes):
             codes = iter(return_codes)
             try:
@@ -139,7 +141,8 @@
             except BaseException, err:
                 return err
             return None
-        event = spawn(accept, [201,500])
+
+        event = spawn(accept, [201, 500])
         for dev in cu.get_container_ring().devs:
             if dev is not None:
                 dev['port'] = bindsock.getsockname()[1]
@@ -155,5 +158,6 @@
             raise err
         self.assert_(not os.path.exists(op_path))
 
+
 if __name__ == '__main__':
     unittest.main()

=== modified file 'test/unit/proxy/test_server.py'
--- test/unit/proxy/test_server.py	2010-11-05 14:47:43 +0000
+++ test/unit/proxy/test_server.py	2010-11-08 18:51:48 +0000
@@ -34,7 +34,7 @@
 from eventlet import sleep, spawn, TimeoutError, util, wsgi, listen
 from eventlet.timeout import Timeout
 import simplejson
-from webob import Request
+from webob import Request, Response
 from webob.exc import HTTPUnauthorized
 
 from test.unit import connect_tcp, readuntil2crlfs
@@ -44,7 +44,7 @@
 from swift.obj import server as object_server
 from swift.common import ring
 from swift.common.constraints import MAX_META_NAME_LENGTH, \
-    MAX_META_VALUE_LENGTH, MAX_META_COUNT, MAX_META_OVERALL_SIZE, MAX_FILE_SIZE
+    MAX_META_VALUE_LENGTH, MAX_META_COUNT, MAX_META_OVERALL_SIZE
 from swift.common.utils import mkdirs, normalize_timestamp, NullLogger
 
 
@@ -53,7 +53,9 @@
 
 
 def fake_http_connect(*code_iter, **kwargs):
+
     class FakeConn(object):
+
         def __init__(self, status, etag=None, body=''):
             self.status = status
             self.reason = 'Fake'
@@ -158,6 +160,7 @@
 
 
 class FakeMemcache(object):
+
     def __init__(self):
         self.store = {}
 
@@ -212,9 +215,12 @@
 class TestProxyServer(unittest.TestCase):
 
     def test_unhandled_exception(self):
+
         class MyApp(proxy_server.Application):
+
             def get_controller(self, path):
                 raise Exception('this shouldnt be caught')
+
         app = MyApp(None, FakeMemcache(), account_ring=FakeRing(),
                 container_ring=FakeRing(), object_ring=FakeRing())
         req = Request.blank('/account', environ={'REQUEST_METHOD': 'HEAD'})
@@ -323,8 +329,11 @@
             test_status_map((200, 200, 204, 500, 404), 503)
 
     def test_PUT_connect_exceptions(self):
+
         def mock_http_connect(*code_iter, **kwargs):
+
             class FakeConn(object):
+
                 def __init__(self, status):
                     self.status = status
                     self.reason = 'Fake'
@@ -372,8 +381,11 @@
             test_status_map((200, 200, 503, 503, -1), 503)
 
     def test_PUT_send_exceptions(self):
+
         def mock_http_connect(*code_iter, **kwargs):
+
             class FakeConn(object):
+
                 def __init__(self, status):
                     self.status = status
                     self.reason = 'Fake'
@@ -430,15 +442,18 @@
             controller = proxy_server.ObjectController(self.app, 'account',
                 'container', 'object')
             req = Request.blank('/a/c/o', {}, headers={
-                'Content-Length': str(MAX_FILE_SIZE + 1),
+                'Content-Length': str(self.app.max_object_size + 1),
                 'Content-Type': 'foo/bar'})
             self.app.update_request(req)
             res = controller.PUT(req)
             self.assertEquals(res.status_int, 413)
 
     def test_PUT_getresponse_exceptions(self):
+
         def mock_http_connect(*code_iter, **kwargs):
+
             class FakeConn(object):
+
                 def __init__(self, status):
                     self.status = status
                     self.reason = 'Fake'
@@ -633,6 +648,7 @@
                 dev['port'] = 1
 
             class SlowBody():
+
                 def __init__(self):
                     self.sent = 0
 
@@ -680,6 +696,7 @@
                 dev['port'] = 1
 
             class SlowBody():
+
                 def __init__(self):
                     self.sent = 0
 
@@ -1334,7 +1351,9 @@
 
     def test_chunked_put(self):
         # quick test of chunked put w/o PATH_TO_TEST_XFS
+
         class ChunkedFile():
+
             def __init__(self, bytes):
                 self.bytes = bytes
                 self.read_bytes = 0
@@ -1375,12 +1394,13 @@
             req.body_file = ChunkedFile(11)
             self.app.memcache.store = {}
             self.app.update_request(req)
+            orig_max_object_size = self.app.max_object_size
             try:
-                server.MAX_FILE_SIZE = 10
+                self.app.max_object_size = 10
                 res = controller.PUT(req)
                 self.assertEquals(res.status_int, 413)
             finally:
-                server.MAX_FILE_SIZE = MAX_FILE_SIZE
+                self.app.max_object_size = orig_max_object_size
 
     def test_chunked_put_and_a_bit_more(self):
         # Since we're starting up a lot here, we're going to test more than
@@ -1495,6 +1515,7 @@
                 self.assertEquals(headers[:len(exp)], exp)
                 # Check unhandled exception
                 orig_update_request = prosrv.update_request
+
                 def broken_update_request(env, req):
                     raise Exception('fake')
                 prosrv.update_request = broken_update_request
@@ -1545,6 +1566,7 @@
                 # in a test for logging x-forwarded-for (first entry only).
 
                 class Logger(object):
+
                     def info(self, msg):
                         self.msg = msg
                 orig_logger = prosrv.logger
@@ -1568,6 +1590,7 @@
                 # Turn on header logging.
 
                 class Logger(object):
+
                     def info(self, msg):
                         self.msg = msg
                 orig_logger = prosrv.logger
@@ -1919,6 +1942,188 @@
             res = controller.PUT(req)
         self.assert_(called[0])
 
+    def test_GETorHEAD1(self):
+        """
+        Ensures we call GETorHEAD_base again without a range if we do a range
+        request and get a 416 Requested Range Not Satisfiable, just in case the
+        primary object is a manifest.
+        """
+        called_without_range = [False]
+        controller = proxy_server.ObjectController(self.app, 'account',
+                        'container', 'object')
+
+        def local_GETorHEAD_base(req, server_type, partition, nodes, path,
+                                 attempts):
+            if req.range:
+                return Response(status=416)
+            called_without_range[0] = True
+            return Response()
+
+        controller.GETorHEAD_base = local_GETorHEAD_base
+        req = Request.blank('/a/c/o', environ={'REQUEST_METHOD': 'GET'},
+                            headers={'Range': 'bytes=0-1'})
+        controller.GETorHEAD(req)
+        self.assert_(called_without_range[0])
+
+    def test_GETorHEAD2(self):
+        """
+        Ensures we call GETorHEAD_base again if the first request was a HEAD
+        and the primary object is a manifest.
+        """
+        called_get = [False]
+        controller = proxy_server.ObjectController(self.app, 'account',
+                        'container', 'object')
+
+        def local_GETorHEAD_base(req, server_type, partition, nodes, path,
+                                 attempts):
+            if req.method == 'HEAD':
+                return Response(headers={'x-object-type': 'manifest'})
+            elif req.method == 'GET':
+                called_get[0] = True
+                return Response(headers={'x-object-type': 'manifest'},
+                    body=pickle.dumps({'content-length': 0, 'x-segment-size': 1,
+                    'x-timestamp': normalize_timestamp(2),
+                    'etag': 'd41d8cd98f00b204e9800998ecf8427e',
+                    'content-type': 'text/plain'}))
+            else:
+                raise Exception('Unexpected method %s' % req.method)
+
+        controller.GETorHEAD_base = local_GETorHEAD_base
+        req = Request.blank('/a/c/o', environ={'REQUEST_METHOD': 'HEAD'})
+        controller.GETorHEAD(req)
+        self.assert_(called_get[0])
+
+    def test_GETorHEAD3(self):
+        """
+        Ensures we call GETorHEAD_base again without a range if the first
+        request was a GET with range that succeeded and the primary object is a
+        manifest.
+        """
+        called_without_range = [False]
+        controller = proxy_server.ObjectController(self.app, 'account',
+                        'container', 'object')
+
+        def local_GETorHEAD_base(req, server_type, partition, nodes, path,
+                                 attempts):
+            if not req.range:
+                called_without_range[0] = True
+            return Response(headers={'x-object-type': 'manifest'},
+                body=pickle.dumps({'content-length': 0, 'x-segment-size': 1,
+                'x-timestamp': normalize_timestamp(2),
+                'etag': 'd41d8cd98f00b204e9800998ecf8427e',
+                'content-type': 'text/plain'}))
+
+        controller.GETorHEAD_base = local_GETorHEAD_base
+        req = Request.blank('/a/c/o', environ={'REQUEST_METHOD': 'GET'},
+                            headers={'Range': 'bytes=0-1'})
+        controller.GETorHEAD(req)
+        self.assert_(called_without_range[0])
+
+    def test_PUT_segmented_object1(self):
+        with save_globals():
+            proxy_server.http_connect = \
+                fake_http_connect(200, 200, # account, container checks
+                    201, 201, 201, # segment 0
+                    201, 201, 201, # segment 1
+                    201, 201, 201, # segment 2
+                    201, 201, 201) # manifest
+            controller = proxy_server.ObjectController(self.app, 'account',
+                            'container', 'object')
+            self.app.segment_size = 2
+            req = Request.blank('/a/c/o', environ={'REQUEST_METHOD': 'PUT'},
+                                body='12345')
+            resp = controller.PUT(req)
+            self.assertEquals(resp.status_int, 201)
+            self.assertEquals(req.bytes_transferred, 5)
+
+    def test_PUT_segmented_object2(self):
+        """ Same as 1, just with a chunky data source. """
+        with save_globals():
+            proxy_server.http_connect = \
+                fake_http_connect(200, 200, # account, container checks
+                    201, 201, 201, # segment 0
+                    201, 201, 201, # segment 1
+                    201, 201, 201, # segment 2
+                    201, 201, 201) # manifest
+            controller = proxy_server.ObjectController(self.app, 'account',
+                            'container', 'object')
+            self.app.segment_size = 2
+
+            class ChunkedReader(object):
+                
+                def __init__(self):
+                    self.chunk = 0
+
+                def read(self, size):
+                    self.chunk += 1
+                    if self.chunk == 1:
+                        return '123'
+                    elif self.chunk == 2:
+                        return '45'
+                    else:
+                        return ''
+
+            req = Request.blank('/a/c/o', environ={'REQUEST_METHOD': 'PUT',
+                'wsgi.input': ChunkedReader()}, headers={'Content-Length': '5'})
+            resp = controller.PUT(req)
+            self.assertEquals(resp.status_int, 201)
+            self.assertEquals(req.bytes_transferred, 5)
+
+    def test_PUT_segmented_object3(self):
+        """ Failed segment PUT. """
+        with save_globals():
+            proxy_server.http_connect = \
+                fake_http_connect(200, 200, # account, container checks
+                    201, 201, 201, # segment 0
+                    201, 500, 500, # segment 1
+                    201, 201, 201, # segment 2
+                    201, 201, 201) # manifest
+            controller = proxy_server.ObjectController(self.app, 'account',
+                            'container', 'object')
+            self.app.segment_size = 2
+            req = Request.blank('/a/c/o', environ={'REQUEST_METHOD': 'PUT'},
+                                body='12345')
+            resp = controller.PUT(req)
+            self.assertEquals(resp.status_int, 503)
+            self.assertEquals(resp.body.strip(),
+                              'Unable to complete very large file operation.')
+
+    def test_PUT_segmented_object4(self):
+        """ Non-matching etag sent. """
+        with save_globals():
+            proxy_server.http_connect = \
+                fake_http_connect(200, 200, # account, container checks
+                    201, 201, 201, # segment 0
+                    201, 201, 201, # segment 1
+                    201, 201, 201, # segment 2
+                    201, 201, 201) # manifest
+            controller = proxy_server.ObjectController(self.app, 'account',
+                            'container', 'object')
+            self.app.segment_size = 2
+            req = Request.blank('/a/c/o', environ={'REQUEST_METHOD': 'PUT'},
+                                body='12345', headers={'ETag': 'abc'})
+            resp = controller.PUT(req)
+            self.assertEquals(resp.status_int, 422)
+
+    def test_PUT_segmented_object5(self):
+        """ Failed manifest PUT. """
+        with save_globals():
+            proxy_server.http_connect = \
+                fake_http_connect(200, 200, # account, container checks
+                    201, 201, 201, # segment 0
+                    201, 201, 201, # segment 1
+                    201, 201, 201, # segment 2
+                    201, 500, 500) # manifest
+            controller = proxy_server.ObjectController(self.app, 'account',
+                            'container', 'object')
+            self.app.segment_size = 2
+            req = Request.blank('/a/c/o', environ={'REQUEST_METHOD': 'PUT'},
+                                body='12345')
+            resp = controller.PUT(req)
+            self.assertEquals(resp.status_int, 503)
+            self.assertEquals(resp.body.strip(),
+                              'Unable to complete very large file operation.')
+
     def test_COPY_calls_authorize(self):
         called = [False]
 
@@ -2080,7 +2285,9 @@
                 self.assertEquals(resp.status_int, 404)
 
     def test_put_locking(self):
+
         class MockMemcache(FakeMemcache):
+
             def __init__(self, allow_lock=None):
                 self.allow_lock = allow_lock
                 super(MockMemcache, self).__init__()
@@ -2669,5 +2876,142 @@
             self.assertEquals(resp.status_int, 400)
 
 
+class TestSegmentedIterable(unittest.TestCase):
+
+    def setUp(self):
+        self.app = proxy_server.Application(None, FakeMemcache(),
+            account_ring=FakeRing(), container_ring=FakeRing(),
+            object_ring=FakeRing())
+        self.controller = proxy_server.ObjectController(self.app, 'account',
+            'container', 'object')
+
+    def test_zero_bytes(self):
+        si = proxy_server.SegmentedIterable(self.controller, 0, 2,
+                                            normalize_timestamp(1))
+        self.assertEquals(''.join(iter(si)), '')
+
+    def test_happy_path(self):
+        segment = [0]
+
+        def give_connect(*args, **kwargs):
+            self.assertEquals(int(kwargs['headers']['X-Object-Segment']),
+                              segment[0])
+            segment[0] += 1
+
+        with save_globals():
+            proxy_server.http_connect = fake_http_connect(200, 200, 200,
+                body='12', give_connect=give_connect)
+            si = proxy_server.SegmentedIterable(self.controller, 5, 2,
+                                                normalize_timestamp(1))
+            self.assertEquals(''.join(iter(si)), '12121')
+            self.assertEquals(segment[0], 3)
+
+    def test_not_found_start(self):
+        with save_globals():
+            proxy_server.http_connect = \
+                fake_http_connect(404, 404, 404, 200, 200, 200, body='12')
+            si = proxy_server.SegmentedIterable(self.controller, 5, 2,
+                                                normalize_timestamp(1))
+            exc = None
+            try:
+                for chunk in si:
+                    raise Exception('Got data when we should not have.')
+            except Exception, err:
+                exc = err
+            self.assertEquals(str(exc),
+                'Could not load segment 0 of /account/container/object')
+
+    def test_not_found_after_start(self):
+        with save_globals():
+            proxy_server.http_connect = \
+                fake_http_connect(200, 404, 404, 404, 200, 200, body='12')
+            si = proxy_server.SegmentedIterable(self.controller, 5, 2,
+                                                normalize_timestamp(1))
+            exc = None
+            try:
+                for chunk in si:
+                    self.assertEquals(chunk, '12')
+            except Exception, err:
+                exc = err
+            self.assertEquals(str(exc),
+                'Could not load segment 1 of /account/container/object')
+
+    def test_partial_not_found(self):
+        with save_globals():
+            proxy_server.http_connect = \
+                fake_http_connect(404, 200, 404, 404, 200, 200, body='12')
+            si = proxy_server.SegmentedIterable(self.controller, 5, 2,
+                                                normalize_timestamp(1))
+            self.assertEquals(''.join(iter(si)), '12121')
+
+    def test_bytes_transferred(self):
+        with save_globals():
+            proxy_server.http_connect = \
+                fake_http_connect(200, 200, 200, body='12')
+            si = proxy_server.SegmentedIterable(self.controller, 5, 2,
+                                                normalize_timestamp(1))
+
+            class Stub(object):
+                pass
+
+            si.response = Stub()
+            self.assertEquals(''.join(iter(si)), '12121')
+            self.assertEquals(si.response.bytes_transferred, 5)
+
+    def test_bytes_transferred_app_iter_range(self):
+        with save_globals():
+            proxy_server.http_connect = \
+                fake_http_connect(200, 200, 200, body='12')
+            si = proxy_server.SegmentedIterable(self.controller, 5, 2,
+                                                normalize_timestamp(1))
+
+            class Stub(object):
+                pass
+
+            si.response = Stub()
+            self.assertEquals(''.join(si.app_iter_range(1, 3)), '212')
+            self.assertEquals(si.response.bytes_transferred, 3)
+
+    def test_app_iter_range_past_end(self):
+        with save_globals():
+            proxy_server.http_connect = \
+                fake_http_connect(200, 200, 200, body='12')
+            si = proxy_server.SegmentedIterable(self.controller, 5, 2,
+                                                normalize_timestamp(1))
+            self.assertEquals(''.join(si.app_iter_range(1, 30)), '2121')
+
+    def test_app_iter_range_start_past_end(self):
+        with save_globals():
+            proxy_server.http_connect = \
+                fake_http_connect(200, 200, 200, body='12')
+            si = proxy_server.SegmentedIterable(self.controller, 5, 2,
+                                                normalize_timestamp(1))
+            self.assertEquals(''.join(si.app_iter_range(30, 31)), '')
+
+    def test_app_iter_range_to_end(self):
+        with save_globals():
+            proxy_server.http_connect = \
+                fake_http_connect(200, 200, 200, body='12')
+            si = proxy_server.SegmentedIterable(self.controller, 5, 2,
+                                                normalize_timestamp(1))
+            self.assertEquals(''.join(si.app_iter_range(3, None)), '21')
+
+    def test_app_iter_range_to_an_end(self):
+        with save_globals():
+            proxy_server.http_connect = \
+                fake_http_connect(200, 200, 200, body='12')
+            si = proxy_server.SegmentedIterable(self.controller, 5, 2,
+                                                normalize_timestamp(1))
+            self.assertEquals(''.join(si.app_iter_range(None, 3)), '121')
+
+    def test_app_iter_range_full(self):
+        with save_globals():
+            proxy_server.http_connect = \
+                fake_http_connect(200, 200, 200, body='12')
+            si = proxy_server.SegmentedIterable(self.controller, 5, 2,
+                                                normalize_timestamp(1))
+            self.assertEquals(''.join(si.app_iter_range(None, None)), '12121')
+
+
 if __name__ == '__main__':
     unittest.main()