[ARVADOS] updated: aff4a730ad890564ee05c2395c4ebb49458e3cdc
git at public.curoverse.com
git at public.curoverse.com
Thu Jun 11 13:12:18 EDT 2015
Summary of changes:
.../app/assets/javascripts/infinite_scroll.js | 8 +-
apps/workbench/app/assets/javascripts/tab_panes.js | 18 +-
.../app/controllers/collections_controller.rb | 17 +-
.../app/controllers/projects_controller.rb | 11 +-
apps/workbench/app/models/arvados_api_client.rb | 14 +-
apps/workbench/app/models/arvados_base.rb | 23 +-
.../collections/_show_source_summary.html.erb | 13 +-
apps/workbench/app/views/layouts/body.html.erb | 6 +
apps/workbench/app/views/projects/public.html.erb | 29 ++
apps/workbench/config/application.default.yml | 6 +-
apps/workbench/config/routes.rb | 2 +
.../controllers/collections_controller_test.rb | 7 +
.../test/controllers/projects_controller_test.rb | 32 ++
apps/workbench/test/helpers/manifest_examples.rb | 1 +
apps/workbench/test/helpers/time_block.rb | 1 +
.../test/integration/anonymous_access_test.rb | 5 +
.../test/integration/application_layout_test.rb | 2 +
.../collection_unit_test.rb | 71 +++++
.../collections_controller_test.rb | 71 +++++
.../collections_perf_test.rb | 116 +++++++
apps/workbench/test/performance/browsing_test.rb | 2 +
apps/workbench/test/test_helper.rb | 34 ++-
apps/workbench/test/unit/arvados_base_test.rb | 87 ++++++
doc/_config.yml | 3 +-
doc/_includes/_arv_run_redirection.liquid | 2 +
doc/_includes/_ssh_addkey.liquid | 15 +-
.../_tutorial_bwa_sortsam_pipeline.liquid | 15 +-
doc/_includes/_tutorial_cluster_name.liquid | 3 +
.../_tutorial_expectations_workstation.liquid | 3 +
doc/_layouts/default.html.liquid | 1 +
doc/css/button-override.css | 7 +
doc/index.html.liquid | 7 +-
.../create-standard-objects.html.textile.liquid | 2 +-
.../install-compute-node.html.textile.liquid | 109 +++++++
.../install-crunch-dispatch.html.textile.liquid | 70 ++++-
...l-manual-prerequisites-ruby.html.textile.liquid | 3 +-
.../getting_started/community.html.textile.liquid | 4 +-
.../getting_started/workbench.html.textile.liquid | 2 +-
doc/user/index.html.textile.liquid | 2 +-
doc/user/reference/api-tokens.html.textile.liquid | 4 +-
doc/user/topics/arv-docker.html.textile.liquid | 4 +-
doc/user/topics/arv-run.html.textile.liquid | 6 +
doc/user/topics/arv-web.html.textile.liquid | 3 +
doc/user/topics/run-command.html.textile.liquid | 2 +
...nning-pipeline-command-line.html.textile.liquid | 10 +-
doc/user/topics/tutorial-job1.html.textile.liquid | 8 +-
.../topics/tutorial-parallel.html.textile.liquid | 2 +-
.../running-external-program.html.textile.liquid | 17 +-
.../tutorial-firstscript.html.textile.liquid | 10 +-
.../tutorial-keep-get.html.textile.liquid | 2 +-
.../tutorials/tutorial-keep.html.textile.liquid | 11 +-
...tutorial-pipeline-workbench.html.textile.liquid | 8 +-
.../tutorial-submit-job.html.textile.liquid | 8 +-
docker/api/Dockerfile | 4 +-
docker/api/apache2_vhost.in | 2 -
docker/api/application.yml.in | 11 +-
docker/api/arvados-clients.yml.in | 2 +-
.../{keep_server_1.json => keep_server_0.json.in} | 3 +-
.../{keep_server_0.json => keep_server_1.json.in} | 2 +-
docker/api/omniauth.rb.in | 2 +-
docker/api/setup-gitolite.sh.in | 4 +-
docker/arvdock | 105 +++++--
docker/base/Dockerfile | 2 +-
docker/build_tools/Makefile | 11 +-
docker/build_tools/build.rb | 52 ++--
docker/config.yml.example | 12 +-
docker/doc/Dockerfile | 5 +-
docker/doc/{apache2_vhost => apache2_vhost.in} | 5 +-
docker/sso/apache2_vhost.in | 2 -
docker/sso/database.yml.in | 22 ++
docker/workbench/Dockerfile | 4 +-
docker/workbench/apache2_vhost.in | 13 +-
docker/workbench/application.yml.in | 6 +-
sdk/cli/bin/crunch-job | 340 ++++++++++++++-------
sdk/go/arvadosclient/arvadosclient.go | 33 +-
sdk/go/arvadosclient/arvadosclient_test.go | 14 +
sdk/go/keepclient/keepclient.go | 19 +-
sdk/go/keepclient/keepclient_test.go | 115 ++++++-
sdk/go/keepclient/support.go | 11 +-
sdk/java/pom.xml | 12 +-
.../java/org/arvados/sdk/{java => }/Arvados.java | 24 +-
.../org/arvados/sdk/{java => }/MethodDetails.java | 4 +-
sdk/python/arvados/api.py | 21 ++
sdk/python/arvados/arvfile.py | 325 ++++++++++++--------
sdk/python/arvados/collection.py | 86 +++---
sdk/python/arvados/commands/arv_copy.py | 5 +-
sdk/python/arvados/commands/keepdocker.py | 148 ++++++---
sdk/python/arvados/keep.py | 25 +-
sdk/python/setup.py | 3 +-
sdk/python/tests/arvados_testutil.py | 4 +-
sdk/python/tests/run_test_server.py | 9 +
sdk/python/tests/test_api.py | 18 ++
sdk/python/tests/test_arvfile.py | 181 ++++++-----
sdk/python/tests/test_keep_client.py | 10 +
.../api/app/controllers/application_controller.rb | 13 +-
services/api/app/models/arvados_model.rb | 21 +-
services/api/app/models/blob.rb | 15 +-
services/api/app/models/collection.rb | 12 +-
services/api/app/models/keep_service.rb | 1 +
services/api/app/models/node.rb | 86 ++++--
services/api/config/application.default.yml | 53 ++--
.../20150512193020_read_only_on_keep_services.rb | 5 +
...50526180251_leading_space_on_full_text_index.rb | 41 +++
services/api/db/structure.sql | 24 +-
.../arvados/v1/collections_controller_test.rb | 38 ++-
services/api/test/helpers/manifest_examples.rb | 31 ++
services/api/test/helpers/time_block.rb | 11 +
.../integration/collections_performance_test.rb | 40 +++
.../api/test/unit/collection_performance_test.rb | 61 ++++
services/api/test/unit/node_test.rb | 49 +++
services/keepproxy/keepproxy_test.go | 8 +-
services/keepstore/bufferpool.go | 6 +-
services/keepstore/bufferpool_test.go | 8 +-
services/keepstore/handler_test.go | 79 +++++
services/keepstore/handlers.go | 6 -
services/keepstore/pull_worker.go | 2 +-
116 files changed, 2453 insertions(+), 752 deletions(-)
create mode 100644 apps/workbench/app/views/projects/public.html.erb
create mode 120000 apps/workbench/test/helpers/manifest_examples.rb
create mode 120000 apps/workbench/test/helpers/time_block.rb
create mode 100644 apps/workbench/test/integration_performance/collection_unit_test.rb
create mode 100644 apps/workbench/test/integration_performance/collections_controller_test.rb
create mode 100644 apps/workbench/test/integration_performance/collections_perf_test.rb
create mode 100644 apps/workbench/test/unit/arvados_base_test.rb
create mode 100644 doc/_includes/_tutorial_cluster_name.liquid
create mode 100644 doc/_includes/_tutorial_expectations_workstation.liquid
create mode 100644 doc/css/button-override.css
create mode 100644 doc/install/install-compute-node.html.textile.liquid
rename docker/api/{keep_server_1.json => keep_server_0.json.in} (59%)
rename docker/api/{keep_server_0.json => keep_server_1.json.in} (59%)
rename docker/doc/{apache2_vhost => apache2_vhost.in} (62%)
create mode 100644 docker/sso/database.yml.in
rename sdk/java/src/main/java/org/arvados/sdk/{java => }/Arvados.java (96%)
rename sdk/java/src/main/java/org/arvados/sdk/{java => }/MethodDetails.java (94%)
create mode 100644 services/api/db/migrate/20150512193020_read_only_on_keep_services.rb
create mode 100644 services/api/db/migrate/20150526180251_leading_space_on_full_text_index.rb
create mode 100644 services/api/test/helpers/manifest_examples.rb
create mode 100644 services/api/test/helpers/time_block.rb
create mode 100644 services/api/test/integration/collections_performance_test.rb
create mode 100644 services/api/test/unit/collection_performance_test.rb
via aff4a730ad890564ee05c2395c4ebb49458e3cdc (commit)
via f9b617b7c8245d1e0eedaafc181501a6ac344657 (commit)
via 900b548097c68649ae2874ded5849f1d8164384c (commit)
via 3d067159ab3cdc2fe3e16c4626a62144d9983968 (commit)
via 2f3cb5d486bfd399f759296905a9d42872e33eaa (commit)
via efc42e74d68cd131db98a2582077bd34d72ebc0b (commit)
via 1c03f0c13bd55683a12c68d9bfc2a602b7815a16 (commit)
via e8ae9364dce380d305833f35dfb25578175664d7 (commit)
via 1b7af451ac13a5813594ec10296bedc894d98a59 (commit)
via 456967e1991ea8adc30038b60f5c34703b47b694 (commit)
via 6a6cd4516f0753f8bbe15265f299e83860ff7174 (commit)
via a9c1ba7bffe05b9231102dcad33546b39a59e823 (commit)
via 18e24e8b11b92909ddeafe81d3f64ad85f9cb097 (commit)
via e56061a95228f9c81085cdf30a68c60931cdaf66 (commit)
via 833c4e9022ab3efd6a8f9277f7e0d281250bb636 (commit)
via 511daa2d275143d89600f015ee0bb19dcbe5641c (commit)
via b33597e94eec5f21fbbcfa72611a3b36397755d5 (commit)
via 869f3f0ad541c85bfb72060b1018905d3f882292 (commit)
via a12cf91db3ececed782d9718af30553f8866ad81 (commit)
via 5881feb407af2d4e5785ad57232cc3bfd12b86ec (commit)
via a341306ea30917b0b5ac245259bd0380c497e4ca (commit)
via 7cd190acfc0b6ce7f65b3effa1e2ce75ed692a3d (commit)
via 0d540b1fde91b85db18fa027de65bb67c5389477 (commit)
via b193c49962deb916893e1ecb0ab04df8b00e3d7a (commit)
via ebdaedbc54c80730733c61cecb3998e26cf5ee7b (commit)
via 375e0f68744fdd73f13921e8449de8c6fb232169 (commit)
via 55f5486de7676b8906066b290ed0420b19f90eda (commit)
via 24e15ff4b9d357d59827db9ec4f1bd165086eecb (commit)
via 4e27d97b0161e56943a5828262a0fb873b826f22 (commit)
via 602706ef5510b3f07fc5fa988019952d2133320c (commit)
via 39a1340d56f7acbddb771f6bef36b68ee9076885 (commit)
via c9e40c6e552bc35f661eb3ed8eff04786cf40566 (commit)
via 22c8b6367a9cd79b17240b7dca1ac8f7d8e7ee77 (commit)
via de8324b3fbbf3f67f0f61c162f5895e8dcd3142d (commit)
via 69f592e029493afb8a0709811b5be1fefabafb4b (commit)
via 4004979fcf1572336a86660b783fcdbebf658db7 (commit)
via c9f0347802b2d7b4844f283072c7931504e25ade (commit)
via 95a1e7bbc1d6f931a0ea50cef9ceb1ea5074a76c (commit)
via c4c8977ef25cc6805f2cca1dedfc83faecc0bc23 (commit)
via de1e5fd5605aaf11b96ef411201e11ac767fe8ba (commit)
via d9ab8c81c11120c32864858d7caafe908c408ad5 (commit)
via 1da9a2a61d66601ab9a02bff439d610ee19c5932 (commit)
via ac4a24f999b9c87ca5ecf6fa9c72204e11a89e66 (commit)
via 214ad0d556ff3e0a7d6cef45cd8e84917994dab6 (commit)
via b5da9565dd4f27394a65ed321f15ee1c3f8ec2e3 (commit)
via 78ddad37d72c6c3a728530dc6932fb91f7d81b87 (commit)
via 7ca020894d276edf9098132f4757cdd46b6b1441 (commit)
via 4b3b3064b87a07b2ba8035dd5c8f3660dd3b2a67 (commit)
via 7a0a91fa4e59e1712611c5d52953898417b50038 (commit)
via a0ee198f06282e0e5fb6325bb4de852d08546eaf (commit)
via baac22ee9ae8f27a10df875e5f1e17b1b6cd51c8 (commit)
via c766dd2394de3480be2047f4c073e5802a001d07 (commit)
via f645742ae8131314f8d5cbd70f10504fc5fd1101 (commit)
via 52e32bbee4cb43cc42aca47c927643f7662da266 (commit)
via 469356d1a60754381e33736ac4f80e1a1e593a7f (commit)
via 121625abcf70672531b35dc4092a4597d8eca4be (commit)
via 76e42f169c6e278c1d8cefe9fb7c03cc70892bac (commit)
via c6fe632f972e4610ff7f35f83a5d7dcd2d6e7ecb (commit)
via 178d3f36265e0e9e9cc0bb6ac8c7c47a9c701687 (commit)
via 299c01cb4d4a96a94ece77db417ceb7af8e9ba69 (commit)
via 9413eb733015601af699f2027d9a7a5bad3f3dea (commit)
via b25916b93d720f80b5bdf16f018d83c13c6c3001 (commit)
via 1ec1d552c77e18e2912e400ae395ca00f4e51c3c (commit)
via 1e0d770315a7b01c316b6a4c314bff58856bfe02 (commit)
via 5e0dd2c6b8f080c81ff6077e629f5ec9f377802f (commit)
via 7a53d874994a5a9af273cee1329d9635b7e03edb (commit)
via 1b5c30eb957594e00a09df745df7630f661e3807 (commit)
via fa9fc5ab6440415542badc8bee0b144d698ec5cc (commit)
via 32038d56fbfe5b635b1c53f247aea9abcca2285c (commit)
via c51a3888a01543d0835119574960a02fd7d35994 (commit)
via 3101fea587bc08e0f4f00a583d1d9e2c953417c1 (commit)
via 6fed84983e6e973eefff66a126f4bb7811c44d29 (commit)
via e73222a8d0c18b159ae3d8b53b54474650bdda16 (commit)
via d9214af111bb44deed1246ec97624b6bd8d970c5 (commit)
via 59d68bda41fcc83c47795a19dbe1d0c180952a08 (commit)
via 15009c109bee16bc09c9bbc11f7df0a677a0cf23 (commit)
via a2eb98fa68672d6966233d7864c328feb8df3939 (commit)
via b3e2782f72e0ca381f3f8c508227ebd7a2ef0c92 (commit)
via 53b19718f974e7c9014644ce80fa36363ae0b693 (commit)
via 2578f9c122d5c4d17cabf9d7f374f14773324360 (commit)
via 7a47332fa99627c3ed0e22a02c54afcb6d128ef5 (commit)
via 6bc89efda1447c479642c7459d14a7afa440019d (commit)
via 9b67268ff45d2a552d21fa39f5086180f537ab4a (commit)
via d3fcdc7afcaf32886d2b492e136f1790e8e93ce4 (commit)
via 872cc3d79d99f5fc4b26b970c3d269ca363ea27c (commit)
via 79564b0ac7d03327cc351bbd6df544ab1f776380 (commit)
via 8a2a84af560d9de68cf0d0c9e34e0d1f19b65e43 (commit)
via 0f125dd51c7dc13047672dee5362866f31885e0a (commit)
via 1a0ba80d44b4be962f898ee1458faea9d4a59137 (commit)
via 71f1ccba7367ca9d000e53c0b86c0448600350f4 (commit)
via 80d57cdcc9fcae93da2b507942815d9de6dd3351 (commit)
via fbcc5ab5e7a5ad65b81e3965f93b1679d1b1e06f (commit)
via b4c2a96ed66c98cb5a07bd06138f755ae5675fb7 (commit)
via a7d558d6c2db242a6555c7f397cb4d91618aa13c (commit)
via 3bf22ae161deadd56d20c4165d2ec569de2dcdef (commit)
via 8b658d96a7bb087406fee97ee4ba9feb830abfd4 (commit)
via 1a2d8a738cf017c7b3f3234a1ea1aa90abb8162e (commit)
via 5d26f7e8bf84d5ed055dc1e88996a25c4db80e85 (commit)
via ae3bb0033a24a38489c49ffc26e5a5e8fd93c160 (commit)
via 9b44bba316583b68f326920d378fbbc77b4567a2 (commit)
via 59129909453452693bd6e7be144cb9417479e98f (commit)
via bf0c6769acfd8728c1e7563f26eb6f5e1f589874 (commit)
via 5151ef795b2ac294137842eea181ab054c76e4a9 (commit)
via c6dd02273d5288f6c6436224e636bb81aba581ca (commit)
via e5c66d135d453a4af4c978590b99a118f65ae12b (commit)
via 35c8b0a58073ab45912d794c90a4c40c17a371be (commit)
via d144724e1b48fa10e28f1e0ddddf440c6ac90ee2 (commit)
via a5401e349b25eb0464b06fd291e9c1a18fb494aa (commit)
via 45d885d56aa967e9699f5245fea30e71f1e2c47e (commit)
via 6114400831cdcd851be517acc436d46555ced966 (commit)
via 82f7fcd2f8f57c197dda487285897cbb92a2da21 (commit)
via 974ab38a740e0dbde9d28ac8848a864981a177e5 (commit)
via 0f6a00174a83252ff1d72ded61289abf5b34a417 (commit)
via f3c4565205413dfeeaa1d2f58634250497d4cf71 (commit)
via 836999ec3a4d5fc6c9f087a03b131bab06121079 (commit)
via 3c517a48e838cbd1c8b1fcc27623fb9ddd33c356 (commit)
via 29009a1c1f8a9653042c5853832881aca4141cf2 (commit)
via 4110970a34689ed526c3365e9b64a784fba7efda (commit)
via d5142a69848df7fa506b8cb16a76cb621598769a (commit)
via b085f051e459981a385c95c4ae147546312c08d9 (commit)
via f1827e2044aff826e63826880b296a59c4a17e2a (commit)
via 100a504ea7f42cc29e71800150cf95c90efe6f67 (commit)
via 28e6503a62fbbff41224843c8c39f4052242c8cd (commit)
via 6702cdabbecef52d342c6f8c67c22cd15377165c (commit)
via e41260f6a5b14753fcad27fbd0e0515a051a4671 (commit)
via d39be4ad92ae92f72367d633f92208f8487ba1b2 (commit)
via 47b10ff0fa5416a5c9fadfd73bb69b1817aac1be (commit)
via 1f2a1c4d4e12109291578a70e1a1316960343eeb (commit)
via 8eb65b21a7b7eea1f987d7fec9cb8ac65de2aeb2 (commit)
via a709514c6bc4e171541106e908316f01ab6fdb60 (commit)
via a007e4eb0b9f00bcd2203261bbfba4aec7aa59bc (commit)
via 92a203c4cf8713a7e28e3d18117d167878dbf804 (commit)
via 9e20ed8567fd8f622f02ad1fc43d486dc3a601b1 (commit)
via b7c7f20d17afc11d2cce7bfee2205a26dc4a4762 (commit)
via 7a7b8ee61a48c40c45f3c600ca782d81a57f27a1 (commit)
via 46b2b9208c6f20c54938ca2312fa56a7f3147472 (commit)
via 0aaa4681a5ad936aeaa08b9b30b167d9f7cac7cc (commit)
via af934d0ffc1d26efc766bb18c226dae70c2179b7 (commit)
via 6b0931644aa30ad9d271ea06c8fa413696c562c0 (commit)
via a88e87137911142ef7e9237703a1073e4feffdb7 (commit)
via 731547ae2d548cd2869779c200813262c05581b2 (commit)
via 37fe7a30a22b669818a185df137c34da09022af3 (commit)
via 77f11805c9daf93356ad083f03841b546aab9c7d (commit)
via 60460bc7b4a04c04aec915221f808c3577343460 (commit)
via 311dc34b42e16f26b82bdf168dc11b39557dac19 (commit)
via d14dd75b263d8f999603b66d23f74667d36a2412 (commit)
via 17fb6bca5786bad5cff83a187e70adbca9c1d14a (commit)
via 7e78bad94194f9ce7a39888cbac30eaf4fbf4af7 (commit)
via acf8433929b812913208e0868599d3a507c7cf34 (commit)
via bfee777f81cbdb447a107f47d6077994e93e4b8c (commit)
via 7538d084fa8778289053f62a6fb8d3ea04868258 (commit)
via e5f039f2e451790ff3838c5810a22a0749d18f6d (commit)
via 37ab170974e6b15c961605b8ed51678ed58a3a52 (commit)
from 19c43cebdfa1b701b031577f62c2a7b060dc8b0e (commit)
Those revisions listed above that are new to this repository have
not appeared on any other notification email; so we list those
revisions in full, below.
commit aff4a730ad890564ee05c2395c4ebb49458e3cdc
Author: Peter Amstutz <peter.amstutz at curoverse.com>
Date: Thu Jun 11 13:12:21 2015 -0400
3198: arvfile.py and collection.py updates based on feedback.
* Add retryable ERROR state to buffer blocks.
* Fix race conditions in testing a bufferblock state before trying to change state.
* Change "modified" to "committed" (which has the opposite meaning, but is more
accurate the way the flag is used)
* Refactor block manager background upload/download threads
* Fix Collection.mkdirs() to match behavior of os.makedirs()
diff --git a/sdk/python/arvados/arvfile.py b/sdk/python/arvados/arvfile.py
index 553cc15..5befa19 100644
--- a/sdk/python/arvados/arvfile.py
+++ b/sdk/python/arvados/arvfile.py
@@ -236,6 +236,13 @@ def synchronized(orig_func):
return orig_func(self, *args, **kwargs)
return synchronized_wrapper
+
+class StateChangeError(Exception):
+ def __init__(self, message, state, nextstate):
+ super(StateChangeError, self).__init__(message)
+ self.state = state
+ self.nextstate = nextstate
+
class _BufferBlock(object):
"""A stand-in for a Keep block that is in the process of being written.
@@ -259,6 +266,7 @@ class _BufferBlock(object):
WRITABLE = 0
PENDING = 1
COMMITTED = 2
+ ERROR = 3
def __init__(self, blockid, starting_capacity, owner):
"""
@@ -280,6 +288,8 @@ class _BufferBlock(object):
self._locator = None
self.owner = owner
self.lock = threading.Lock()
+ self.wait_for_commit = threading.Event()
+ self.error = None
@synchronized
def append(self, data):
@@ -302,16 +312,27 @@ class _BufferBlock(object):
raise AssertionError("Buffer block is not writable")
@synchronized
- def set_state(self, nextstate, loc=None):
+ def set_state(self, nextstate, val=None):
if ((self._state == _BufferBlock.WRITABLE and nextstate == _BufferBlock.PENDING) or
- (self._state == _BufferBlock.PENDING and nextstate == _BufferBlock.COMMITTED)):
+ (self._state == _BufferBlock.PENDING and nextstate == _BufferBlock.COMMITTED) or
+ (self._state == _BufferBlock.PENDING and nextstate == _BufferBlock.ERROR) or
+ (self._state == _BufferBlock.ERROR and nextstate == _BufferBlock.PENDING)):
self._state = nextstate
+
+ if self._state == _BufferBlock.PENDING:
+ self.wait_for_commit.clear()
+
if self._state == _BufferBlock.COMMITTED:
- self._locator = loc
+ self._locator = val
self.buffer_view = None
self.buffer_block = None
+ self.wait_for_commit.set()
+
+ if self._state == _BufferBlock.ERROR:
+ self.error = val
+ self.wait_for_commit.set()
else:
- raise AssertionError("Invalid state change from %s to %s" % (self.state, nextstate))
+ raise StateChangeError("Invalid state change from %s to %s" % (self.state, nextstate), self.state, nextstate)
@synchronized
def state(self):
@@ -331,7 +352,7 @@ class _BufferBlock(object):
@synchronized
def clone(self, new_blockid, owner):
if self._state == _BufferBlock.COMMITTED:
- raise AssertionError("Can only duplicate a writable or pending buffer block")
+ raise AssertionError("Cannot duplicate committed buffer block")
bufferblock = _BufferBlock(new_blockid, self.size(), owner)
bufferblock.append(self.buffer_view[0:self.size()])
return bufferblock
@@ -373,19 +394,22 @@ class _BlockManager(object):
Collection of ArvadosFiles.
"""
+
+ DEFAULT_PUT_THREADS = 2
+ DEFAULT_GET_THREADS = 2
+
def __init__(self, keep):
"""keep: KeepClient object to use"""
self._keep = keep
self._bufferblocks = {}
self._put_queue = None
- self._put_errors = None
self._put_threads = None
self._prefetch_queue = None
self._prefetch_threads = None
self.lock = threading.Lock()
self.prefetch_enabled = True
- self.num_put_threads = 2
- self.num_get_threads = 2
+ self.num_put_threads = _BlockManager.DEFAULT_PUT_THREADS
+ self.num_get_threads = _BlockManager.DEFAULT_GET_THREADS
@synchronized
def alloc_bufferblock(self, blockid=None, starting_capacity=2**14, owner=None):
@@ -427,6 +451,70 @@ class _BlockManager(object):
def is_bufferblock(self, locator):
return locator in self._bufferblocks
+ def _commit_bufferblock_worker(self):
+ """Background uploader thread."""
+
+ while True:
+ try:
+ bufferblock = self._put_queue.get()
+ if bufferblock is None:
+ return
+
+ loc = self._keep.put(bufferblock.buffer_view[0:bufferblock.write_pointer].tobytes())
+ bufferblock.set_state(_BufferBlock.COMMITTED, loc)
+
+ except Exception as e:
+ bufferblock.set_state(_BufferBlock.ERROR, e)
+ finally:
+ if self._put_queue is not None:
+ self._put_queue.task_done()
+
+ @synchronized
+ def start_put_threads(self):
+ if self._put_threads is None:
+ # Start uploader threads.
+
+ # If we don't limit the Queue size, the upload queue can quickly
+ # grow to take up gigabytes of RAM if the writing process is
+ # generating data more quickly than it can be send to the Keep
+ # servers.
+ #
+ # With two upload threads and a queue size of 2, this means up to 4
+ # blocks pending. If they are full 64 MiB blocks, that means up to
+ # 256 MiB of internal buffering, which is the same size as the
+ # default download block cache in KeepClient.
+ self._put_queue = Queue.Queue(maxsize=2)
+
+ self._put_threads = []
+ for i in xrange(0, self.num_put_threads):
+ thread = threading.Thread(target=self._commit_bufferblock_worker)
+ self._put_threads.append(thread)
+ thread.daemon = False
+ thread.start()
+
+ def _block_prefetch_worker(self):
+ """The background downloader thread."""
+ while True:
+ try:
+ b = self._prefetch_queue.get()
+ if b is None:
+ return
+ self._keep.get(b)
+ except Exception:
+ pass
+
+ @synchronized
+ def start_get_threads(self):
+ if self._prefetch_threads is None:
+ self._prefetch_queue = Queue.Queue()
+ self._prefetch_threads = []
+ for i in xrange(0, self.num_get_threads):
+ thread = threading.Thread(target=self._block_prefetch_worker)
+ self._prefetch_threads.append(thread)
+ thread.daemon = True
+ thread.start()
+
+
@synchronized
def stop_threads(self):
"""Shut down and wait for background upload and download threads to finish."""
@@ -438,7 +526,6 @@ class _BlockManager(object):
t.join()
self._put_threads = None
self._put_queue = None
- self._put_errors = None
if self._prefetch_threads is not None:
for t in self._prefetch_threads:
@@ -448,71 +535,48 @@ class _BlockManager(object):
self._prefetch_threads = None
self._prefetch_queue = None
- def commit_bufferblock(self, block, wait):
+ def __enter__(self):
+ return self
+
+ def __exit__(self, exc_type, exc_value, traceback):
+ self.stop_threads()
+
+ def __del__(self):
+ self.stop_threads()
+
+ def commit_bufferblock(self, block, sync):
"""Initiate a background upload of a bufferblock.
:block:
The block object to upload
- :wait:
- If `wait` is True, upload the block synchronously.
- If `wait` is False, upload the block asynchronously. This will
+ :sync:
+ If `sync` is True, upload the block synchronously.
+ If `sync` is False, upload the block asynchronously. This will
return immediately unless if the upload queue is at capacity, in
which case it will wait on an upload queue slot.
"""
- def commit_bufferblock_worker(self):
- """Background uploader thread."""
-
- while True:
- try:
- bufferblock = self._put_queue.get()
- if bufferblock is None:
- return
-
- loc = self._keep.put(bufferblock.buffer_view[0:bufferblock.write_pointer].tobytes())
- bufferblock.set_state(_BufferBlock.COMMITTED, loc)
-
- except Exception as e:
- self._put_errors.put((bufferblock.locator(), e))
- finally:
- if self._put_queue is not None:
- self._put_queue.task_done()
-
- if block.state() != _BufferBlock.WRITABLE:
+ try:
+ # Mark the block as PENDING so to disallow any more appends.
+ block.set_state(_BufferBlock.PENDING)
+ except StateChangeError as e:
+ if e.state == _BufferBlock.PENDING and sync:
+ block.wait_for_commit.wait()
+ if block.state() == _BufferBlock.ERROR:
+ raise block.error
return
- if wait:
- block.set_state(_BufferBlock.PENDING)
- loc = self._keep.put(block.buffer_view[0:block.write_pointer].tobytes())
- block.set_state(_BufferBlock.COMMITTED, loc)
+ if sync:
+ try:
+ loc = self._keep.put(block.buffer_view[0:block.write_pointer].tobytes())
+ block.set_state(_BufferBlock.COMMITTED, loc)
+ except Exception as e:
+ block.set_state(_BufferBlock.ERROR, e)
+ raise
else:
- with self.lock:
- if self._put_threads is None:
- # Start uploader threads.
-
- # If we don't limit the Queue size, the upload queue can quickly
- # grow to take up gigabytes of RAM if the writing process is
- # generating data more quickly than it can be send to the Keep
- # servers.
- #
- # With two upload threads and a queue size of 2, this means up to 4
- # blocks pending. If they are full 64 MiB blocks, that means up to
- # 256 MiB of internal buffering, which is the same size as the
- # default download block cache in KeepClient.
- self._put_queue = Queue.Queue(maxsize=2)
- self._put_errors = Queue.Queue()
-
- self._put_threads = []
- for i in xrange(0, self.num_put_threads):
- thread = threading.Thread(target=commit_bufferblock_worker, args=(self,))
- self._put_threads.append(thread)
- thread.daemon = True
- thread.start()
-
- # Mark the block as PENDING so to disallow any more appends.
- block.set_state(_BufferBlock.PENDING)
+ self.start_put_threads()
self._put_queue.put(block)
@synchronized
@@ -547,36 +611,33 @@ class _BlockManager(object):
def commit_all(self):
"""Commit all outstanding buffer blocks.
- Unlike commit_bufferblock(), this is a synchronous call, and will not
- return until all buffer blocks are uploaded. Raises
- KeepWriteError() if any blocks failed to upload.
+ This is a synchronous call, and will not return until all buffer blocks
+ are uploaded. Raises KeepWriteError() if any blocks failed to upload.
"""
with self.lock:
items = self._bufferblocks.items()
for k,v in items:
- if v.state() == _BufferBlock.WRITABLE:
- v.owner.flush(False)
+ if v.state() != _BufferBlock.COMMITTED:
+ v.owner.flush(sync=False)
with self.lock:
if self._put_queue is not None:
self._put_queue.join()
- if not self._put_errors.empty():
- err = []
- try:
- while True:
- err.append(self._put_errors.get(False))
- except Queue.Empty:
- pass
+ err = []
+ for k,v in items:
+ if v.state() == _BufferBlock.ERROR:
+ err.append((v.locator(), v.error))
+ if err:
raise KeepWriteError("Error writing some blocks", err, label="block")
for k,v in items:
- # flush again with wait=True to remove committed bufferblocks from
+ # flush again with sync=True to remove committed bufferblocks from
# the segments.
if v.owner:
- v.owner.flush(True)
+ v.owner.flush(sync=True)
def block_prefetch(self, locator):
@@ -592,28 +653,10 @@ class _BlockManager(object):
if not self.prefetch_enabled:
return
- def block_prefetch_worker(self):
- """The background downloader thread."""
- while True:
- try:
- b = self._prefetch_queue.get()
- if b is None:
- return
- self._keep.get(b)
- except Exception:
- pass
-
with self.lock:
if locator in self._bufferblocks:
return
- if self._prefetch_threads is None:
- self._prefetch_queue = Queue.Queue()
- self._prefetch_threads = []
- for i in xrange(0, self.num_get_threads):
- thread = threading.Thread(target=block_prefetch_worker, args=(self,))
- self._prefetch_threads.append(thread)
- thread.daemon = True
- thread.start()
+ self.start_get_threads()
self._prefetch_queue.put(locator)
@@ -640,7 +683,7 @@ class ArvadosFile(object):
"""
self.parent = parent
self.name = name
- self._modified = True
+ self._committed = False
self._segments = []
self.lock = parent.root_collection().lock
for s in segments:
@@ -681,7 +724,7 @@ class ArvadosFile(object):
self._segments.append(Range(new_loc, other_segment.range_start, other_segment.range_size, other_segment.segment_offset))
- self._modified = True
+ self._committed = False
def __eq__(self, other):
if other is self:
@@ -717,14 +760,14 @@ class ArvadosFile(object):
return not self.__eq__(other)
@synchronized
- def set_unmodified(self):
- """Clear the modified flag"""
- self._modified = False
+ def set_committed(self):
+ """Set committed flag to False"""
+ self._committed = True
@synchronized
- def modified(self):
- """Test the modified flag"""
- return self._modified
+ def committed(self):
+ """Get whether this is committed or not."""
+ return self._committed
@must_be_writable
@synchronized
@@ -752,7 +795,7 @@ class ArvadosFile(object):
new_segs.append(r)
self._segments = new_segs
- self._modified = True
+ self._committed = False
elif size > self.size():
raise IOError(errno.EINVAL, "truncate() does not support extending the file size")
@@ -833,7 +876,7 @@ class ArvadosFile(object):
n += config.KEEP_BLOCK_SIZE
return
- self._modified = True
+ self._committed = False
if self._current_bblock is None or self._current_bblock.state() != _BufferBlock.WRITABLE:
self._current_bblock = self.parent._my_block_manager().alloc_bufferblock(owner=self)
@@ -841,7 +884,7 @@ class ArvadosFile(object):
if (self._current_bblock.size() + len(data)) > config.KEEP_BLOCK_SIZE:
self._repack_writes(num_retries)
if (self._current_bblock.size() + len(data)) > config.KEEP_BLOCK_SIZE:
- self.parent._my_block_manager().commit_bufferblock(self._current_bblock, False)
+ self.parent._my_block_manager().commit_bufferblock(self._current_bblock, sync=False)
self._current_bblock = self.parent._my_block_manager().alloc_bufferblock(owner=self)
self._current_bblock.append(data)
@@ -853,26 +896,35 @@ class ArvadosFile(object):
return len(data)
@synchronized
- def flush(self, wait=True, num_retries=0):
- """Flush bufferblocks to Keep."""
- if self.modified():
- if self._current_bblock and self._current_bblock.state() == _BufferBlock.WRITABLE:
+ def flush(self, sync=True, num_retries=0):
+ """Flush the current bufferblock to Keep.
+
+ :sync:
+ If True, commit block synchronously, wait until buffer block has been written.
+ If False, commit block asynchronously, return immediately after putting block into
+ the keep put queue.
+ """
+ if self.committed():
+ return
+
+ if self._current_bblock and self._current_bblock.state() != _BufferBlock.COMMITTED:
+ if self._current_bblock.state() == _BufferBlock.WRITABLE:
self._repack_writes(num_retries)
- self.parent._my_block_manager().commit_bufferblock(self._current_bblock, wait)
- if wait:
- to_delete = set()
- for s in self._segments:
- bb = self.parent._my_block_manager().get_bufferblock(s.locator)
- if bb:
- if bb.state() != _BufferBlock.COMMITTED:
- _logger.error("bufferblock %s is not committed" % (s.locator))
- else:
- to_delete.add(s.locator)
- s.locator = bb.locator()
- for s in to_delete:
- self.parent._my_block_manager().delete_bufferblock(s)
-
- self.parent.notify(MOD, self.parent, self.name, (self, self))
+ self.parent._my_block_manager().commit_bufferblock(self._current_bblock, sync=sync)
+
+ if sync:
+ to_delete = set()
+ for s in self._segments:
+ bb = self.parent._my_block_manager().get_bufferblock(s.locator)
+ if bb:
+ if bb.state() != _BufferBlock.COMMITTED:
+ self.parent._my_block_manager().commit_bufferblock(self._current_bblock, sync=True)
+ to_delete.add(s.locator)
+ s.locator = bb.locator()
+ for s in to_delete:
+ self.parent._my_block_manager().delete_bufferblock(s)
+
+ self.parent.notify(MOD, self.parent, self.name, (self, self))
@must_be_writable
@synchronized
@@ -887,7 +939,7 @@ class ArvadosFile(object):
def _add_segment(self, blocks, pos, size):
"""Internal implementation of add_segment."""
- self._modified = True
+ self._committed = False
for lr in locators_and_ranges(blocks, pos, size):
last = self._segments[-1] if self._segments else Range(0, 0, 0, 0)
r = Range(lr.locator, last.range_start+last.range_size, lr.segment_size, lr.segment_offset)
@@ -921,8 +973,8 @@ class ArvadosFile(object):
@must_be_writable
@synchronized
def _reparent(self, newparent, newname):
- self._modified = True
- self.flush()
+ self._committed = False
+ self.flush(sync=True)
self.parent.remove(self.name)
self.parent = newparent
self.name = newname
@@ -937,8 +989,8 @@ class ArvadosFileReader(ArvadosFileReaderBase):
"""
- def __init__(self, arvadosfile, mode="r", num_retries=None):
- super(ArvadosFileReader, self).__init__(arvadosfile.name, mode, num_retries=num_retries)
+ def __init__(self, arvadosfile, num_retries=None):
+ super(ArvadosFileReader, self).__init__(arvadosfile.name, "r", num_retries=num_retries)
self.arvadosfile = arvadosfile
def size(self):
@@ -990,7 +1042,8 @@ class ArvadosFileWriter(ArvadosFileReader):
"""
def __init__(self, arvadosfile, mode, num_retries=None):
- super(ArvadosFileWriter, self).__init__(arvadosfile, mode, num_retries=num_retries)
+ super(ArvadosFileWriter, self).__init__(arvadosfile, num_retries=num_retries)
+ self.mode = mode
@_FileLikeObjectBase._before_close
@retry_method
diff --git a/sdk/python/arvados/collection.py b/sdk/python/arvados/collection.py
index cb090ec..6677ca6 100644
--- a/sdk/python/arvados/collection.py
+++ b/sdk/python/arvados/collection.py
@@ -487,7 +487,7 @@ class RichCollectionBase(CollectionBase):
def __init__(self, parent=None):
self.parent = parent
- self._modified = True
+ self._committed = False
self._callback = None
self._items = {}
@@ -542,7 +542,7 @@ class RichCollectionBase(CollectionBase):
else:
item = ArvadosFile(self, pathcomponents[0])
self._items[pathcomponents[0]] = item
- self._modified = True
+ self._committed = False
self.notify(ADD, self, pathcomponents[0], item)
return item
else:
@@ -550,12 +550,12 @@ class RichCollectionBase(CollectionBase):
# create new collection
item = Subcollection(self, pathcomponents[0])
self._items[pathcomponents[0]] = item
- self._modified = True
+ self._committed = False
self.notify(ADD, self, pathcomponents[0], item)
if isinstance(item, RichCollectionBase):
return item.find_or_create(pathcomponents[1], create_type)
else:
- raise IOError(errno.ENOTDIR, "Is not a directory: %s" % pathcomponents[0])
+ raise IOError(errno.ENOTDIR, "Not a directory: '%s'" % pathcomponents[0])
else:
return self
@@ -583,13 +583,18 @@ class RichCollectionBase(CollectionBase):
else:
raise IOError(errno.ENOTDIR, "Is not a directory: %s" % pathcomponents[0])
+ @synchronized
def mkdirs(self, path):
"""Recursive subcollection create.
- Like `os.mkdirs()`. Will create intermediate subcollections needed to
- contain the leaf subcollection path.
+ Like `os.makedirs()`. Will create intermediate subcollections needed
+ to contain the leaf subcollection path.
"""
+
+ if self.find(path) != None:
+ raise IOError(errno.EEXIST, "Directory or file exists: '%s'" % path)
+
return self.find_or_create(path, COLLECTION)
def open(self, path, mode="r"):
@@ -634,26 +639,32 @@ class RichCollectionBase(CollectionBase):
name = os.path.basename(path)
if mode == "r":
- return ArvadosFileReader(arvfile, mode, num_retries=self.num_retries)
+ return ArvadosFileReader(arvfile, num_retries=self.num_retries)
else:
return ArvadosFileWriter(arvfile, mode, num_retries=self.num_retries)
- @synchronized
def modified(self):
- """Test if the collection (or any subcollection or file) has been modified."""
- if self._modified:
- return True
+ return not self.committed()
+
+ def set_unmodified(self):
+ self.set_committed()
+
+ @synchronized
+ def committed(self):
+ """Test if the collection and all subcollection and files are committed."""
+ if self._committed is False:
+ return False
for v in self._items.values():
- if v.modified():
- return True
- return False
+ if v.committed() is False:
+ return False
+ return True
@synchronized
- def set_unmodified(self):
- """Recursively clear modified flag."""
- self._modified = False
+ def set_committed(self):
+ """Recursively set committed flag."""
+ self._committed = True
for k,v in self._items.items():
- v.set_unmodified()
+ v.set_committed()
@synchronized
def __iter__(self):
@@ -684,7 +695,7 @@ class RichCollectionBase(CollectionBase):
def __delitem__(self, p):
"""Delete an item by name which is directly contained by this collection."""
del self._items[p]
- self._modified = True
+ self._committed = False
self.notify(DEL, self, p, None)
@synchronized
@@ -727,7 +738,7 @@ class RichCollectionBase(CollectionBase):
raise IOError(errno.ENOTEMPTY, "Subcollection not empty")
deleteditem = self._items[pathcomponents[0]]
del self._items[pathcomponents[0]]
- self._modified = True
+ self._committed = False
self.notify(DEL, self, pathcomponents[0], deleteditem)
else:
item.remove(pathcomponents[1])
@@ -776,7 +787,7 @@ class RichCollectionBase(CollectionBase):
item = source_obj.clone(self, target_name)
self._items[target_name] = item
- self._modified = True
+ self._committed = False
if modified_from:
self.notify(MOD, self, target_name, (modified_from, item))
@@ -885,6 +896,7 @@ class RichCollectionBase(CollectionBase):
"""
return self._get_manifest_text(stream_name, True, True)
+ @synchronized
def manifest_text(self, stream_name=".", strip=False, normalize=False):
"""Get the manifest text for this collection, sub collections and files.
@@ -928,7 +940,7 @@ class RichCollectionBase(CollectionBase):
"""
- if self.modified() or self._manifest_text is None or normalize:
+ if not self.committed() or self._manifest_text is None or normalize:
stream = {}
buf = []
sorted_keys = sorted(self.keys())
@@ -989,7 +1001,7 @@ class RichCollectionBase(CollectionBase):
"""
if changes:
- self._modified = True
+ self._committed = False
for change in changes:
event_type = change[0]
path = change[1]
@@ -1377,7 +1389,7 @@ class Collection(RichCollectionBase):
Retry count on API calls (if None, use the collection default)
"""
- if self.modified():
+ if not self.committed():
if not self._has_collection_uuid():
raise AssertionError("Collection manifest_locator is not a collection uuid. Use save_new() for new collections.")
@@ -1393,7 +1405,7 @@ class Collection(RichCollectionBase):
).execute(
num_retries=num_retries)
self._manifest_text = self._api_response["manifest_text"]
- self.set_unmodified()
+ self.set_committed()
return self._manifest_text
@@ -1452,7 +1464,7 @@ class Collection(RichCollectionBase):
self._manifest_locator = self._api_response["uuid"]
self._manifest_text = text
- self.set_unmodified()
+ self.set_committed()
return text
@@ -1485,7 +1497,7 @@ class Collection(RichCollectionBase):
segments = []
streamoffset = 0L
state = BLOCKS
- self.mkdirs(stream_name)
+ self.find_or_create(stream_name, COLLECTION)
continue
if state == BLOCKS:
@@ -1511,13 +1523,13 @@ class Collection(RichCollectionBase):
raise errors.SyntaxError("File %s conflicts with stream of the same name.", filepath)
else:
# error!
- raise errors.SyntaxError("Invalid manifest format")
+ raise errors.SyntaxError("Invalid manifest format, expected file segment but did not match format: '%s'" % tok)
if sep == "\n":
stream_name = None
state = STREAM_NAME
- self.set_unmodified()
+ self.set_committed()
@synchronized
def notify(self, event, collection, name, item):
@@ -1567,7 +1579,7 @@ class Subcollection(RichCollectionBase):
@must_be_writable
@synchronized
def _reparent(self, newparent, newname):
- self._modified = True
+ self._committed = False
self.flush()
self.parent.remove(self.name, recursive=True)
self.parent = newparent
diff --git a/sdk/python/tests/test_arvfile.py b/sdk/python/tests/test_arvfile.py
index a01543a..99be4c2 100644
--- a/sdk/python/tests/test_arvfile.py
+++ b/sdk/python/tests/test_arvfile.py
@@ -77,11 +77,11 @@ class ArvadosFileWriterTestCase(unittest.TestCase):
writer.seek(0, os.SEEK_SET)
self.assertEqual("01234567", writer.read(12))
- self.assertEqual(None, c.manifest_locator())
- self.assertEqual(True, c.modified())
+ self.assertIsNone(c.manifest_locator())
+ self.assertTrue(c.modified())
c.save_new("test_truncate")
self.assertEqual("zzzzz-4zz18-mockcollection0", c.manifest_locator())
- self.assertEqual(False, c.modified())
+ self.assertFalse(c.modified())
def test_write_to_end(self):
keep = ArvadosFileWriterTestCase.MockKeep({"781e5e245d69b566979b86e28d23f2c7+10": "0123456789"})
@@ -104,13 +104,13 @@ class ArvadosFileWriterTestCase(unittest.TestCase):
writer.seek(5, os.SEEK_SET)
self.assertEqual("56789foo", writer.read(8))
- self.assertEqual(None, c.manifest_locator())
- self.assertEqual(True, c.modified())
- self.assertEqual(None, keep.get("acbd18db4cc2f85cedef654fccc4a4d8+3"))
+ self.assertIsNone(c.manifest_locator())
+ self.assertTrue(c.modified())
+ self.assertIsNone(keep.get("acbd18db4cc2f85cedef654fccc4a4d8+3"))
c.save_new("test_append")
self.assertEqual("zzzzz-4zz18-mockcollection0", c.manifest_locator())
- self.assertEqual(False, c.modified())
+ self.assertFalse(c.modified())
self.assertEqual("foo", keep.get("acbd18db4cc2f85cedef654fccc4a4d8+3"))
@@ -233,11 +233,11 @@ class ArvadosFileWriterTestCase(unittest.TestCase):
writer.write(text)
self.assertEqual(writer.size(), 100000000)
- self.assertEqual(None, c.manifest_locator())
- self.assertEqual(True, c.modified())
+ self.assertIsNone(c.manifest_locator())
+ self.assertTrue(c.modified())
c.save_new("test_write_large")
self.assertEqual("zzzzz-4zz18-mockcollection0", c.manifest_locator())
- self.assertEqual(False, c.modified())
+ self.assertFalse(c.modified())
def test_large_write(self):
@@ -326,11 +326,11 @@ class ArvadosFileWriterTestCase(unittest.TestCase):
writer.write("foo")
self.assertEqual(writer.size(), 100000000)
- self.assertEqual(None, c.manifest_locator())
- self.assertEqual(True, c.modified())
+ self.assertIsNone(c.manifest_locator())
+ self.assertTrue(c.modified())
c.save_new("test_write_large")
self.assertEqual("zzzzz-4zz18-mockcollection0", c.manifest_locator())
- self.assertEqual(False, c.modified())
+ self.assertFalse(c.modified())
def test_create(self):
keep = ArvadosFileWriterTestCase.MockKeep({})
@@ -344,12 +344,12 @@ class ArvadosFileWriterTestCase(unittest.TestCase):
writer.write("01234567")
self.assertEqual(writer.size(), 8)
- self.assertEqual(None, c.manifest_locator())
- self.assertEqual(True, c.modified())
- self.assertEqual(None, keep.get("2e9ec317e197819358fbc43afca7d837+8"))
+ self.assertIsNone(c.manifest_locator())
+ self.assertTrue(c.modified())
+ self.assertIsNone(keep.get("2e9ec317e197819358fbc43afca7d837+8"))
c.save_new("test_create")
self.assertEqual("zzzzz-4zz18-mockcollection0", c.manifest_locator())
- self.assertEqual(False, c.modified())
+ self.assertFalse(c.modified())
self.assertEqual("01234567", keep.get("2e9ec317e197819358fbc43afca7d837+8"))
@@ -363,7 +363,9 @@ class ArvadosFileWriterTestCase(unittest.TestCase):
self.assertIsNone(c.api_response())
writer = c.open("foo/bar/count.txt", "w+")
writer.write("01234567")
+ self.assertFalse(c.committed())
c.save_new("test_create")
+ self.assertTrue(c.committed())
self.assertEqual(c.api_response(), api.response)
def test_overwrite(self):
@@ -379,11 +381,11 @@ class ArvadosFileWriterTestCase(unittest.TestCase):
writer.write("01234567")
self.assertEqual(writer.size(), 8)
- self.assertEqual(None, c.manifest_locator())
- self.assertEqual(True, c.modified())
+ self.assertIsNone(c.manifest_locator())
+ self.assertTrue(c.modified())
c.save_new("test_overwrite")
self.assertEqual("zzzzz-4zz18-mockcollection0", c.manifest_locator())
- self.assertEqual(False, c.modified())
+ self.assertFalse(c.modified())
def test_file_not_found(self):
with Collection('. 781e5e245d69b566979b86e28d23f2c7+10 0:10:count.txt\n') as c:
@@ -409,12 +411,12 @@ class ArvadosFileWriterTestCase(unittest.TestCase):
self.assertEqual(w1.size(), 8)
self.assertEqual(w2.size(), 8)
- self.assertEqual(None, c.manifest_locator())
- self.assertEqual(True, c.modified())
- self.assertEqual(None, keep.get("2e9ec317e197819358fbc43afca7d837+8"))
+ self.assertIsNone(c.manifest_locator())
+ self.assertTrue(c.modified())
+ self.assertIsNone(keep.get("2e9ec317e197819358fbc43afca7d837+8"))
c.save_new("test_create_multiple")
self.assertEqual("zzzzz-4zz18-mockcollection0", c.manifest_locator())
- self.assertEqual(False, c.modified())
+ self.assertFalse(c.modified())
self.assertEqual("01234567", keep.get("2e9ec317e197819358fbc43afca7d837+8"))
@@ -558,81 +560,85 @@ class ArvadosFileReadlinesTestCase(ArvadosFileReadTestCase):
class BlockManagerTest(unittest.TestCase):
def test_bufferblock_append(self):
keep = ArvadosFileWriterTestCase.MockKeep({})
- blockmanager = arvados.arvfile._BlockManager(keep)
- bufferblock = blockmanager.alloc_bufferblock()
- bufferblock.append("foo")
+ with arvados.arvfile._BlockManager(keep) as blockmanager:
+ bufferblock = blockmanager.alloc_bufferblock()
+ bufferblock.append("foo")
- self.assertEqual(bufferblock.size(), 3)
- self.assertEqual(bufferblock.buffer_view[0:3], "foo")
- self.assertEqual(bufferblock.locator(), "acbd18db4cc2f85cedef654fccc4a4d8+3")
+ self.assertEqual(bufferblock.size(), 3)
+ self.assertEqual(bufferblock.buffer_view[0:3], "foo")
+ self.assertEqual(bufferblock.locator(), "acbd18db4cc2f85cedef654fccc4a4d8+3")
- bufferblock.append("bar")
+ bufferblock.append("bar")
- self.assertEqual(bufferblock.size(), 6)
- self.assertEqual(bufferblock.buffer_view[0:6], "foobar")
- self.assertEqual(bufferblock.locator(), "3858f62230ac3c915f300c664312c63f+6")
+ self.assertEqual(bufferblock.size(), 6)
+ self.assertEqual(bufferblock.buffer_view[0:6], "foobar")
+ self.assertEqual(bufferblock.locator(), "3858f62230ac3c915f300c664312c63f+6")
- bufferblock.set_state(arvados.arvfile._BufferBlock.PENDING)
- with self.assertRaises(arvados.errors.AssertionError):
- bufferblock.append("bar")
+ bufferblock.set_state(arvados.arvfile._BufferBlock.PENDING)
+ with self.assertRaises(arvados.errors.AssertionError):
+ bufferblock.append("bar")
def test_bufferblock_dup(self):
keep = ArvadosFileWriterTestCase.MockKeep({})
- blockmanager = arvados.arvfile._BlockManager(keep)
- bufferblock = blockmanager.alloc_bufferblock()
- bufferblock.append("foo")
+ with arvados.arvfile._BlockManager(keep) as blockmanager:
+ bufferblock = blockmanager.alloc_bufferblock()
+ bufferblock.append("foo")
- self.assertEqual(bufferblock.size(), 3)
- self.assertEqual(bufferblock.buffer_view[0:3], "foo")
- self.assertEqual(bufferblock.locator(), "acbd18db4cc2f85cedef654fccc4a4d8+3")
- bufferblock.set_state(arvados.arvfile._BufferBlock.PENDING)
+ self.assertEqual(bufferblock.size(), 3)
+ self.assertEqual(bufferblock.buffer_view[0:3], "foo")
+ self.assertEqual(bufferblock.locator(), "acbd18db4cc2f85cedef654fccc4a4d8+3")
+ bufferblock.set_state(arvados.arvfile._BufferBlock.PENDING)
- bufferblock2 = blockmanager.dup_block(bufferblock, None)
- self.assertNotEqual(bufferblock.blockid, bufferblock2.blockid)
+ bufferblock2 = blockmanager.dup_block(bufferblock, None)
+ self.assertNotEqual(bufferblock.blockid, bufferblock2.blockid)
- bufferblock2.append("bar")
+ bufferblock2.append("bar")
- self.assertEqual(bufferblock2.size(), 6)
- self.assertEqual(bufferblock2.buffer_view[0:6], "foobar")
- self.assertEqual(bufferblock2.locator(), "3858f62230ac3c915f300c664312c63f+6")
+ self.assertEqual(bufferblock2.size(), 6)
+ self.assertEqual(bufferblock2.buffer_view[0:6], "foobar")
+ self.assertEqual(bufferblock2.locator(), "3858f62230ac3c915f300c664312c63f+6")
- self.assertEqual(bufferblock.size(), 3)
- self.assertEqual(bufferblock.buffer_view[0:3], "foo")
- self.assertEqual(bufferblock.locator(), "acbd18db4cc2f85cedef654fccc4a4d8+3")
+ self.assertEqual(bufferblock.size(), 3)
+ self.assertEqual(bufferblock.buffer_view[0:3], "foo")
+ self.assertEqual(bufferblock.locator(), "acbd18db4cc2f85cedef654fccc4a4d8+3")
def test_bufferblock_get(self):
keep = ArvadosFileWriterTestCase.MockKeep({"781e5e245d69b566979b86e28d23f2c7+10": "0123456789"})
- blockmanager = arvados.arvfile._BlockManager(keep)
- bufferblock = blockmanager.alloc_bufferblock()
- bufferblock.append("foo")
+ with arvados.arvfile._BlockManager(keep) as blockmanager:
+ bufferblock = blockmanager.alloc_bufferblock()
+ bufferblock.append("foo")
- self.assertEqual(blockmanager.get_block_contents("781e5e245d69b566979b86e28d23f2c7+10", 1), "0123456789")
- self.assertEqual(blockmanager.get_block_contents(bufferblock.blockid, 1), "foo")
+ self.assertEqual(blockmanager.get_block_contents("781e5e245d69b566979b86e28d23f2c7+10", 1), "0123456789")
+ self.assertEqual(blockmanager.get_block_contents(bufferblock.blockid, 1), "foo")
def test_bufferblock_commit(self):
mockkeep = mock.MagicMock()
- blockmanager = arvados.arvfile._BlockManager(mockkeep)
- bufferblock = blockmanager.alloc_bufferblock()
- bufferblock.owner = mock.MagicMock()
- bufferblock.owner.flush.side_effect = lambda x: blockmanager.commit_bufferblock(bufferblock, False)
- bufferblock.append("foo")
- blockmanager.commit_all()
- self.assertTrue(bufferblock.owner.flush.called)
- self.assertTrue(mockkeep.put.called)
- self.assertEqual(bufferblock.state(), arvados.arvfile._BufferBlock.COMMITTED)
- self.assertIsNone(bufferblock.buffer_view)
+ with arvados.arvfile._BlockManager(mockkeep) as blockmanager:
+ bufferblock = blockmanager.alloc_bufferblock()
+ bufferblock.owner = mock.MagicMock()
+ def flush(sync=None):
+ blockmanager.commit_bufferblock(bufferblock, sync)
+ bufferblock.owner.flush.side_effect = flush
+ bufferblock.append("foo")
+ blockmanager.commit_all()
+ self.assertTrue(bufferblock.owner.flush.called)
+ self.assertTrue(mockkeep.put.called)
+ self.assertEqual(bufferblock.state(), arvados.arvfile._BufferBlock.COMMITTED)
+ self.assertIsNone(bufferblock.buffer_view)
def test_bufferblock_commit_with_error(self):
mockkeep = mock.MagicMock()
mockkeep.put.side_effect = arvados.errors.KeepWriteError("fail")
- blockmanager = arvados.arvfile._BlockManager(mockkeep)
- bufferblock = blockmanager.alloc_bufferblock()
- bufferblock.owner = mock.MagicMock()
- bufferblock.owner.flush.side_effect = lambda x: blockmanager.commit_bufferblock(bufferblock, False)
- bufferblock.append("foo")
- with self.assertRaises(arvados.errors.KeepWriteError) as err:
- blockmanager.commit_all()
- self.assertTrue(bufferblock.owner.flush.called)
- self.assertEqual(str(err.exception), "Error writing some blocks: block acbd18db4cc2f85cedef654fccc4a4d8+3 raised KeepWriteError (fail)")
- self.assertEqual(bufferblock.state(), arvados.arvfile._BufferBlock.PENDING)
+ with arvados.arvfile._BlockManager(mockkeep) as blockmanager:
+ bufferblock = blockmanager.alloc_bufferblock()
+ bufferblock.owner = mock.MagicMock()
+ def flush(sync=None):
+ blockmanager.commit_bufferblock(bufferblock, sync)
+ bufferblock.owner.flush.side_effect = flush
+ bufferblock.append("foo")
+ with self.assertRaises(arvados.errors.KeepWriteError) as err:
+ blockmanager.commit_all()
+ self.assertTrue(bufferblock.owner.flush.called)
+ self.assertEqual(str(err.exception), "Error writing some blocks: block acbd18db4cc2f85cedef654fccc4a4d8+3 raised KeepWriteError (fail)")
+ self.assertEqual(bufferblock.state(), arvados.arvfile._BufferBlock.ERROR)
commit f9b617b7c8245d1e0eedaafc181501a6ac344657
Author: Peter Amstutz <peter.amstutz at curoverse.com>
Date: Wed Jun 10 09:30:07 2015 -0400
3198: Adjust language of errors to say what the error is instead of saying what the
input should be.
diff --git a/sdk/python/arvados/arvfile.py b/sdk/python/arvados/arvfile.py
index 3009a90..553cc15 100644
--- a/sdk/python/arvados/arvfile.py
+++ b/sdk/python/arvados/arvfile.py
@@ -361,7 +361,7 @@ def must_be_writable(orig_func):
@functools.wraps(orig_func)
def must_be_writable_wrapper(self, *args, **kwargs):
if not self.writable():
- raise IOError(errno.EROFS, "Collection must be writable.")
+ raise IOError(errno.EROFS, "Collection is read-only.")
return orig_func(self, *args, **kwargs)
return must_be_writable_wrapper
diff --git a/sdk/python/arvados/collection.py b/sdk/python/arvados/collection.py
index eea0717..cb090ec 100644
--- a/sdk/python/arvados/collection.py
+++ b/sdk/python/arvados/collection.py
@@ -555,7 +555,7 @@ class RichCollectionBase(CollectionBase):
if isinstance(item, RichCollectionBase):
return item.find_or_create(pathcomponents[1], create_type)
else:
- raise IOError(errno.ENOTDIR, "Interior path components must be subcollection")
+ raise IOError(errno.ENOTDIR, "Is not a directory: %s" % pathcomponents[0])
else:
return self
@@ -568,7 +568,7 @@ class RichCollectionBase(CollectionBase):
"""
if not path:
- raise errors.ArgumentError("Parameter 'path' must not be empty.")
+ raise errors.ArgumentError("Parameter 'path' is empty.")
pathcomponents = path.split("/", 1)
item = self._items.get(pathcomponents[0])
@@ -581,7 +581,7 @@ class RichCollectionBase(CollectionBase):
else:
return item
else:
- raise IOError(errno.ENOTDIR, "Interior path components must be subcollection")
+ raise IOError(errno.ENOTDIR, "Is not a directory: %s" % pathcomponents[0])
def mkdirs(self, path):
"""Recursive subcollection create.
@@ -626,7 +626,7 @@ class RichCollectionBase(CollectionBase):
if arvfile is None:
raise IOError(errno.ENOENT, "File not found")
if not isinstance(arvfile, ArvadosFile):
- raise IOError(errno.EISDIR, "Path must refer to a file.")
+ raise IOError(errno.EISDIR, "Is a directory: %s" % path)
if mode[0] == "w":
arvfile.truncate(0)
@@ -716,7 +716,7 @@ class RichCollectionBase(CollectionBase):
"""
if not path:
- raise errors.ArgumentError("Parameter 'path' must not be empty.")
+ raise errors.ArgumentError("Parameter 'path' is empty.")
pathcomponents = path.split("/", 1)
item = self._items.get(pathcomponents[0])
@@ -870,7 +870,7 @@ class RichCollectionBase(CollectionBase):
source_obj, target_dir, target_name = self._get_src_target(source, target_path, source_collection, False)
if not source_obj.writable():
- raise IOError(errno.EROFS, "Source collection must be writable.")
+ raise IOError(errno.EROFS, "Source collection is read only.")
target_dir.add(source_obj, target_name, overwrite, True)
def portable_manifest_text(self, stream_name="."):
@@ -1170,7 +1170,7 @@ class Collection(RichCollectionBase):
self._manifest_text = manifest_locator_or_text
else:
raise errors.ArgumentError(
- "Argument to CollectionReader must be a manifest or a collection UUID")
+ "Argument to CollectionReader is not a manifest or a collection UUID")
try:
self._populate()
@@ -1379,7 +1379,7 @@ class Collection(RichCollectionBase):
"""
if self.modified():
if not self._has_collection_uuid():
- raise AssertionError("Collection manifest_locator must be a collection uuid. Use save_new() for new collections.")
+ raise AssertionError("Collection manifest_locator is not a collection uuid. Use save_new() for new collections.")
self._my_block_manager().commit_all()
diff --git a/sdk/python/arvados/keep.py b/sdk/python/arvados/keep.py
index 1e09172..b2700ae 100644
--- a/sdk/python/arvados/keep.py
+++ b/sdk/python/arvados/keep.py
@@ -76,7 +76,7 @@ class KeepLocator(object):
return getattr(self, data_name)
def setter(self, hex_str):
if not arvados.util.is_hex(hex_str, length):
- raise ValueError("{} must be a {}-digit hex string: {}".
+ raise ValueError("{} is not a {}-digit hex string: {}".
format(name, length, hex_str))
setattr(self, data_name, hex_str)
return property(getter, setter)
@@ -929,7 +929,7 @@ class KeepClient(object):
if isinstance(data, unicode):
data = data.encode("ascii")
elif not isinstance(data, str):
- raise arvados.errors.ArgumentError("Argument 'data' to KeepClient.put must be type 'str'")
+ raise arvados.errors.ArgumentError("Argument 'data' to KeepClient.put is not type 'str'")
data_hash = hashlib.md5(data).hexdigest()
if copies < 1:
commit 900b548097c68649ae2874ded5849f1d8164384c
Merge: 28e6503 3d06715
Author: Peter Amstutz <peter.amstutz at curoverse.com>
Date: Wed Jun 10 09:10:41 2015 -0400
Merge branch 'master' into 3198-writable-fuse
diff --cc sdk/python/arvados/arvfile.py
index 2d44d6a,95dcea0..3009a90
--- a/sdk/python/arvados/arvfile.py
+++ b/sdk/python/arvados/arvfile.py
@@@ -9,9 -9,8 +9,9 @@@ import Queu
import copy
import errno
import re
+import logging
- from .errors import KeepWriteError, AssertionError
+ from .errors import KeepWriteError, AssertionError, ArgumentError
from .keep import KeepLocator
from ._normalize_stream import normalize_stream
from ._ranges import locators_and_ranges, replace_range, Range
commit 28e6503a62fbbff41224843c8c39f4052242c8cd
Merge: 19c43ce e41260f
Author: Peter Amstutz <peter.amstutz at curoverse.com>
Date: Mon May 18 16:50:36 2015 -0400
Merge branch 'master' into 3198-writable-fuse
-----------------------------------------------------------------------
hooks/post-receive
--
More information about the arvados-commits
mailing list