GekkoFS directory and subdirectory working (7c5a9c4e) · Commits · hpc / cargo

lib/libcargo.cpp

+1 −1

Original line number	Diff line number	Diff line
		@@ -307,7 +307,7 @@ transfer::wait() const {
		auto s = status();

		while(!s.done() && !s.failed()) {
		s = wait_for(150ms);
		s = wait_for(1000ms);
		}

		return s;

src/posix_file/posix_file/fs_plugin/gekko_plugin.cpp

+14 −11

Original line number	Diff line number	Diff line
		@@ -79,19 +79,22 @@ gekko_plugin::readdir(const std::string& path) {
		for(auto& file : files) {

		struct stat buf;
		stat("/" + file, &buf);
		std::string correct_path = file;
		if(path.size() != 1) {
		correct_path = path + "/" + file;
		} else {
		correct_path = "/" + file;
		}

		stat(correct_path, &buf);

		if(S_ISDIR(buf.st_mode)) {

		std::vector<std::string> subfiles = readdir("/" + file);
		std::vector<std::string> subfiles = readdir(correct_path);
		final_list.insert(final_list.end(), subfiles.begin(),
		subfiles.end());
		} else {

		if(path.size() != 1) {
		final_list.push_back(path + "/" + file);
		} else {
		final_list.push_back("/" + file);
		}
		final_list.push_back(correct_path);
		}
		}

src/worker/worker.cpp

+13 −11

Original line number	Diff line number	Diff line
		@@ -112,6 +112,7 @@ worker::run() {

		auto I = m_ops.begin();
		auto IE = m_ops.end();
		//LOGGER_INFO ("[Status] Pending: {}", m_ops.size());
		if(I != IE) {
		auto op = I->second.first.get();
		int index = I->second.second;
		@@ -119,9 +120,10 @@ worker::run() {
		if(index == -1) {
		// operation not started
		// Print error message
		update_state(op->source(), op->tid(), op->seqno(),
		op->output_path(), transfer_state::running,
		-1.0f);
		/* We avoid this update, that may not come into order...*/
		//update_state(op->source(), op->tid(), op->seqno(),
		// op->output_path(), transfer_state::running,
		// -1.0f);
		cargo::error_code ec = (*op)();
		if(ec != cargo::error_code::transfer_in_progress) {
		update_state(op->source(), op->tid(), op->seqno(),
		@@ -136,7 +138,7 @@ worker::run() {
		// Operation in progress
		index = op->progress(index);
		if(index == -1) {
		// operation finished
		// operation finishe
		cargo::error_code ec = op->progress();
		update_state(op->source(), op->tid(), op->seqno(),
		op->output_path(),
		@@ -186,6 +188,9 @@ worker::run() {
		for (std::size_t i = 0; i < m.input_path().size(); i++) {
		std::string input_path = m.input_path()[i];
		std::string output_path = m.output_path()[i];
		update_state(msg->source(), m.tid(), i,
		output_path, transfer_state::pending, -1.0f);

		m_ops.emplace(std::make_pair(
		make_pair(input_path, output_path),
		make_pair(operation::make_operation(
		@@ -200,13 +205,10 @@ worker::run() {

		op->set_comm(msg->source(), m.tid(), i, t);

		update_state(op->source(), op->tid(), op->seqno(),
		op->output_path(), transfer_state::pending, -1.0f);
		}

		}
		break;
		}

		case tag::bw_shaping: {
		shaper_message m;
		world.recv(msg->source(), msg->tag(), m);