services/backup/docker-server/contents/server/src/Constants.h
33–34 ↗	(On Diff #10809)	This comment should explain why smaller size allows us to do so. Or it could explain why it is not possible when the size is greater.
services/backup/docker-server/contents/server/src/Reactors/server/SendLogReactor.h
77–82 ↗	(On Diff #10809)	Have you tested this code? We always throw here: if persistenceMethod = PersistenceMethod::UNKNOWN we throw because persistenceMethod != PersistenceMethod::BLOB if persistenceMethod = PersistenceMethod::BLOB we throw because persistenceMethod != PersistenceMethod::UNKNOWN in all the other cases we throw because persistenceMethod != PersistenceMethod::UNKNOWN

This revision now requires changes to proceed.Apr 1 2022, 4:38 AM

• karol added inline comments.Apr 4 2022, 3:43 AM

services/backup/docker-server/contents/server/src/Constants.h
33–34 ↗	(On Diff #10809)	ok
services/backup/docker-server/contents/server/src/Reactors/server/SendLogReactor.h
77–82 ↗	(On Diff #10809)	I saw this issue before, there should be `&&`, not `\|\|`. I thought I updated this but apparently, I did not, thanks for spotting this!

update comment, fix condition

Harbormaster failed remote builds in B7819: Diff 10986!Apr 4 2022, 3:51 AM

tomek requested changes to this revision.Apr 4 2022, 4:21 AM

tomek added inline comments.

services/backup/docker-server/contents/server/src/Constants.h
33–37	Please use sentences (capital letters and dots).
services/backup/docker-server/contents/server/src/Reactors/server/SendLogReactor.h
68–88	I'm not sure if I understand the idea behind this, but this doesn't seem to be correct. Let's suppose we want to persist two chunks, one bigger than `LOG_DATA_SIZE_DATABASE_LIMIT` and one smaller. After the first chunk, we set `this->persistenceMethod = PersistenceMethod::BLOB`. When processing the second, the first `if` condition is true, and we throw because `this->persistenceMethod != PersistenceMethod::UNKNOWN`. Am I missing something? As a side note, it's better to avoid throwing when we can recover from invalid state. In this case, if we started writing to the db and then the size limit is exceeded, we can move the data to blob and write the rest there.

This revision now requires changes to proceed.Apr 4 2022, 4:21 AM

update logic

services/backup/docker-server/contents/server/src/Constants.h
33–37	ok
services/backup/docker-server/contents/server/src/Reactors/server/SendLogReactor.h
68–88	You are right, there's a logic error here, I'm going to fix this. As a side note, it's better to avoid throwing when we can recover from invalid state. In this case, if we started writing to the db and then the size limit is exceeded, we can move the data to blob and write the rest there. This would make sense but I think not in this particular case. The idea here is that storing in DB is for one-chunk operations only. If we ever receive more than one chunk, we should never store the bytes in the DB.

Harbormaster completed remote builds in B7857: Diff 11031.Apr 5 2022, 12:07 AM

tomek requested changes to this revision.Apr 5 2022, 9:35 AM

tomek added inline comments.

services/backup/docker-server/contents/server/src/Reactors/server/SendLogReactor.h
72–80 ↗	(On Diff #11031)	Do we need e.g. `else` here? I guess the intention wasn't to always throw.

This revision now requires changes to proceed.Apr 5 2022, 9:35 AM

• karol added inline comments.Apr 6 2022, 12:49 AM

services/backup/docker-server/contents/server/src/Reactors/server/SendLogReactor.h
72–80 ↗	(On Diff #11031)	Ahh, right!

add else

Harbormaster completed remote builds in B7908: Diff 11087.Apr 6 2022, 12:59 AM

tomek accepted this revision.Apr 6 2022, 6:52 AM

This revision is now accepted and ready to land.Apr 6 2022, 6:52 AM

rebase

Harbormaster completed remote builds in B7946: Diff 11135.Apr 7 2022, 12:34 AM

Closed by commit rCOMMecee8379c2e6: [services] Backup - Send log - Add basic logic (authored by • karol). · Explain WhyApr 7 2022, 2:45 AM

This revision was automatically updated to reflect the committed changes.

• karol added a commit: rCOMMecee8379c2e6: [services] Backup - Send log - Add basic logic.

Revision Contents
Changeset List

Path

Size

services/

backup/

docker-server/

contents/

server/

src/

Constants.h

7 lines

Reactors/

server/

SendLogReactor.h

38 lines

Diff 10986

View Options

services/backup/docker-server/contents/server/src/Constants.h

#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

View Options

services/backup/docker-server/contents/server/src/Reactors/server/SendLogReactor.h

#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

[services] Backup - Send log - Add basic logicClosedPublicActions

Details

Diff Detail

Event Timeline

Revision ContentsChangeset List

Diff 10986

services/backup/docker-server/contents/server/src/Constants.h

services/backup/docker-server/contents/server/src/Reactors/server/SendLogReactor.h

[services] Backup - Send log - Add basic logic
ClosedPublic
Actions

Revision Contents
Changeset List