#pragma once		#pragma once

#include "BlobGetClientReactor.h"		#include "BlobGetClientReactor.h"
		#include "DatabaseEntitiesTools.h"
#include "DatabaseManager.h"		#include "DatabaseManager.h"
#include "ServerWriteReactorBase.h"		#include "ServerWriteReactorBase.h"
#include "ServiceBlobClient.h"		#include "ServiceBlobClient.h"

#include "../_generated/backup.grpc.pb.h"		#include "../_generated/backup.grpc.pb.h"
#include "../_generated/backup.pb.h"		#include "../_generated/backup.pb.h"

#include <folly/MPMCQueue.h>		#include <folly/MPMCQueue.h>

#include <iostream>		#include <iostream>
#include <memory>		#include <memory>
#include <string>		#include <string>
		#include <vector>

namespace comm {		namespace comm {
namespace network {		namespace network {
namespace reactor {		namespace reactor {

class PullBackupReactor : public ServerWriteReactorBase<		class PullBackupReactor : public ServerWriteReactorBase<
backup::PullBackupRequest,		backup::PullBackupRequest,
backup::PullBackupResponse> {		backup::PullBackupResponse> {

enum class State {		enum class State {
COMPACTION = 1,		COMPACTION = 1,
LOGS = 2,		LOGS = 2,
};		};

std::shared_ptr<database::BackupItem> backupItem;		std::shared_ptr<database::BackupItem> backupItem;
std::shared_ptr<reactor::BlobGetClientReactor> getReactor;		std::shared_ptr<reactor::BlobGetClientReactor> getReactor;
std::mutex reactorStateMutex;		std::mutex reactorStateMutex;
std::shared_ptr<folly::MPMCQueue<std::string>> dataChunks;		std::shared_ptr<folly::MPMCQueue<std::string>> dataChunks;
ServiceBlobClient blobClient;		ServiceBlobClient blobClient;
State state = State::COMPACTION;		State state = State::COMPACTION;
		std::vector<std::shared_ptr<database::LogItem>> logs;
		size_t currentLogIndex = 0;
		std::shared_ptr<database::LogItem> currentLog;

void initializeGetReactor(const std::string &holder);		void initializeGetReactor(const std::string &holder);

public:		public:
PullBackupReactor(const backup::PullBackupRequest *request);		PullBackupReactor(const backup::PullBackupRequest *request);

void initialize() override;		void initialize() override;

Show All 34 Lines	void PullBackupReactor::initialize() {
this->backupItem = database::DatabaseManager::getInstance().findBackupItem(		this->backupItem = database::DatabaseManager::getInstance().findBackupItem(
this->request.userid(), this->request.backupid());		this->request.userid(), this->request.backupid());
if (this->backupItem == nullptr) {		if (this->backupItem == nullptr) {
throw std::runtime_error(		throw std::runtime_error(
"no backup found for provided parameters: user id [" +		"no backup found for provided parameters: user id [" +
this->request.userid() + "], backup id [" + this->request.backupid() +		this->request.userid() + "], backup id [" + this->request.backupid() +
"]");		"]");
}		}
// TODO get logs		this->logs = database::DatabaseManager::getInstance().findLogItemsForBackup(
		this->request.backupid());
		tomekUnsubmitted Not Done Inline Actions I remember asking about it somewhere and you've probably answered, but what is the plan for a scenario where someone appends a log after we started reading but before we finish? tomek: I remember asking about it somewhere and you've probably answered, but what is the plan for a…
		karolAuthorUnsubmitted Done Inline Actions I honestly don't remember a discussion like this. What plan could be there? We're going to read the current state of the logs. It feels a bit weird to see a scenario containing two people. Backups are individual, right? karol: I honestly don't remember a discussion like this. What plan could be there? We're going to…
		tomekUnsubmitted Not Done Inline Actions Ok, so let's be more precise. By someone, I meant two different clients, and in that case the same user could be logged in on both. There are three ways of handling this: We can raise an error - we really don't want that We can deliver a backup with a state from the begging of the operation. This is what you suggested. We can deliver a backup with a state from the end of the operation. I think that option 2 is easier, but at the same time we provide something which we know is outdated. Option 3 can be implemented e.g. by checking if there are new entries when we finished sending logs. Option 2 has a big issue when we restore the state from a backup - we might miss some updates that were made after we started the process. We can work-around it by e.g. having `currentAsOf` field and fetch new logs from the client after the backup is restored. So overall, both options are ok, but we should choose one and have it in our minds while implementing the rest of the system. @karol-bisztyga @ashoat tomek: Ok, so let's be more precise. By someone, I meant two different clients, and in that case the…
		ashoatUnsubmitted Not Done Inline Actions The diff was landed without replying to this question! ashoat: The diff was landed without replying to this question!
}		}

std::unique_ptr<grpc::Status>		std::unique_ptr<grpc::Status>
PullBackupReactor::writeResponse(backup::PullBackupResponse *response) {		PullBackupReactor::writeResponse(backup::PullBackupResponse *response) {
// we make sure that the blob client's state is flushed to the main memory		// we make sure that the blob client's state is flushed to the main memory
// as there may be multiple threads from the pool taking over here		// as there may be multiple threads from the pool taking over here
const std::lock_guard<std::mutex> lock(this->reactorStateMutex);		const std::lock_guard<std::mutex> lock(this->reactorStateMutex);
switch (this->state) {		if (this->state == State::COMPACTION) {
case State::COMPACTION: {
this->initializeGetReactor(this->backupItem->getCompactionHolder());		this->initializeGetReactor(this->backupItem->getCompactionHolder());
std::string dataChunk;		std::string dataChunk;
this->dataChunks->blockingRead(dataChunk);		this->dataChunks->blockingRead(dataChunk);
if (dataChunk.empty()) {		if (!dataChunk.empty()) {
// TODO try to immediately start writing logs instead of wasting a cycle		response->set_compactionchunk(dataChunk);
// sending nothing
this->state = State::LOGS;
return nullptr;		return nullptr;
}		}
response->set_compactionchunk(dataChunk);		if (!this->dataChunks->isEmpty()) {
		throw std::runtime_error(
		"dangling data discovered after reading compaction");
		}
		this->getReactor = nullptr;
		this->state = State::LOGS;
		}
		if (this->state == State::LOGS) {
		// TODO make sure logs are received in correct order regardless their size
		if (this->logs.empty()) {
		return std::make_unique<grpc::Status>(grpc::Status::OK);
		}
		if (this->currentLogIndex == this->logs.size()) {
		return std::make_unique<grpc::Status>(grpc::Status::OK);
		}
		if (this->currentLogIndex > this->logs.size()) {
		throw std::runtime_error("log index out of bound");
		}
		if (this->currentLog == nullptr) {
		this->currentLog = this->logs.at(this->currentLogIndex);
		if (this->currentLog->getPersistedInBlob()) {
		this->initializeGetReactor(this->currentLog->getValue());
		} else {
		response->set_logchunk(this->currentLog->getValue());
		++this->currentLogIndex;
		this->currentLog = nullptr;
return nullptr;		return nullptr;
}		}
case State::LOGS: {
throw std::runtime_error("unimplemented");
}		}
		std::string dataChunk;
		this->dataChunks->blockingRead(dataChunk);
		if (dataChunk.empty()) {
		++this->currentLogIndex;
		this->currentLog = nullptr;
		} else {
		response->set_logchunk(dataChunk);
		}
		return nullptr;
}		}
throw std::runtime_error("unhandled state");		throw std::runtime_error("unhandled state");
}		}

} // namespace reactor		} // namespace reactor
} // namespace network		} // namespace network
} // namespace comm		} // namespace comm

Add keyserver/certs to .dockerignoreClosedPublicActions

Details

Diff Detail

Event Timeline

Revision ContentsChangeset List

Diff 44284

.dockerignore

Add keyserver/certs to .dockerignore
ClosedPublic
Actions

Revision Contents
Changeset List