database/pgbackrest
1
0
Fork 0
mirror of https://github.com/pgbackrest/pgbackrest.git synced 2026-10-23 21:28:56 +03:00
Code Activity

Invert storage read/write interface ownership.

Browse Source 
Previously, drivers constructed StorageRead/StorageWrite objects directly and stored metadata in a shared interface struct. Now, StorageRead/StorageWrite create the driver via storageInterfaceNewReadP()/NewWriteP() and mediate between IoRead/IoWrite and the driver. Drivers return opaque objects and own their metadata independently.

This loosens the tight coupling between drivers and the StorageRead/StorageWrite layer. The remote write driver replaces its back-pointer to StorageWrite with a filterGroup callback, eliminating the circular dependency. It makes retry in StorageRead much more readable.

Also move the logic for testing whether a file version could not be found out of the drivers and into StorageRead.
This commit is contained in:
David Steele
2026-06-09 11:23:15 +07:00
parent 71f52f7d92
commit a953aa74b5
47 changed files with 979 additions and 791 deletions
Download Patch File Download Diff File Expand all files Collapse all files

68
src/storage/azure/read.c
View File

@@ -18,32 +18,30 @@ STRING_STATIC(AZURE_QUERY_VERSION_ID_STR, "versionid")
/***********************************************************************************************************************************
Object type
***********************************************************************************************************************************/
typedef struct StorageReadAzure
struct StorageReadAzure
{
StorageReadInterface interface; // Interface
const StorageReadInterface *interface; // Interface
StorageAzure *storage; // Storage that created this object
HttpResponse *httpResponse; // HTTP response
} StorageReadAzure;
const String *name; // File name
uint64_t offset; // Read offset
const Variant *limit; // Read limit (NULL for no limit)
const String *versionId; // Version id (NULL for most recent)
/***********************************************************************************************************************************
Macros for function logging
***********************************************************************************************************************************/
#define FUNCTION_LOG_STORAGE_READ_AZURE_TYPE \
StorageReadAzure *
#define FUNCTION_LOG_STORAGE_READ_AZURE_FORMAT(value, buffer, bufferSize) \
objNameToLog(value, "StorageReadAzure", buffer, bufferSize)
HttpResponse *httpResponse; // HTTP response
};
/***********************************************************************************************************************************
Open the file
***********************************************************************************************************************************/
static bool
storageReadAzureOpen(THIS_VOID)
storageReadAzureOpen(THIS_VOID, const bool ignoreMissing)
{
THIS(StorageReadAzure);
FUNCTION_LOG_BEGIN(logLevelTrace);
FUNCTION_LOG_PARAM(STORAGE_READ_AZURE, this);
FUNCTION_LOG_PARAM(BOOL, ignoreMissing);
FUNCTION_LOG_END();
ASSERT(this != NULL);
@@ -51,30 +49,23 @@ storageReadAzureOpen(THIS_VOID)
bool result = false;
// Read if not versioned or if versionId is not null
if (!this->interface.version || this->interface.versionId != NULL)
MEM_CONTEXT_OBJ_BEGIN(this)
{
// Request the file
MEM_CONTEXT_OBJ_BEGIN(this)
{
this->httpResponse = storageAzureRequestP(
this->storage, HTTP_VERB_GET_STR, .path = this->interface.name,
.query =
this->interface.version ?
httpQueryPut(httpQueryNewP(), AZURE_QUERY_VERSION_ID_STR, this->interface.versionId) : NULL,
.header = httpHeaderPutRange(httpHeaderNew(NULL), this->interface.offset, this->interface.limit),
.allowMissing = true, .contentIo = true);
}
MEM_CONTEXT_OBJ_END();
this->httpResponse = storageAzureRequestP(
this->storage, HTTP_VERB_GET_STR, .path = this->name,
.query = this->versionId != NULL ? httpQueryPut(httpQueryNewP(), AZURE_QUERY_VERSION_ID_STR, this->versionId) : NULL,
.header = httpHeaderPutRange(httpHeaderNew(NULL), this->offset, this->limit), .allowMissing = true, .contentIo = true);
// If file exists
if (httpResponseCodeOk(this->httpResponse))
{
result = true;
}
// Else error unless ignore missing
else if (!this->interface.ignoreMissing)
THROW_FMT(FileMissingError, STORAGE_ERROR_READ_MISSING, strZ(this->interface.name));
else if (!ignoreMissing)
THROW_FMT(FileMissingError, STORAGE_ERROR_READ_MISSING, strZ(this->name));
}
MEM_CONTEXT_OBJ_END();
FUNCTION_LOG_RETURN(BOOL, result);
}
@@ -140,7 +131,7 @@ storageReadAzureClose(THIS_VOID)
}
/**********************************************************************************************************************************/
static const IoReadInterface storageReadAzureInterface =
static const StorageReadInterface storageReadAzureInterface =
{
.close = storageReadAzureClose,
.eof = storageReadAzureEof,
@@ -148,18 +139,16 @@ static const IoReadInterface storageReadAzureInterface =
.read = storageReadAzure,
};
FN_EXTERN StorageRead *
FN_EXTERN StorageReadAzure *
storageReadAzureNew(
StorageAzure *const storage, const String *const name, const bool ignoreMissing, const uint64_t offset,
const Variant *const limit, const bool version, const String *const versionId)
StorageAzure *const storage, const String *const name, const uint64_t offset, const Variant *const limit,
const String *const versionId)
{
FUNCTION_LOG_BEGIN(logLevelTrace);
FUNCTION_LOG_PARAM(STORAGE_AZURE, storage);
FUNCTION_LOG_PARAM(STRING, name);
FUNCTION_LOG_PARAM(BOOL, ignoreMissing);
FUNCTION_LOG_PARAM(UINT64, offset);
FUNCTION_LOG_PARAM(VARIANT, limit);
FUNCTION_LOG_PARAM(BOOL, version);
FUNCTION_LOG_PARAM(STRING, versionId);
FUNCTION_LOG_END();
@@ -170,14 +159,15 @@ storageReadAzureNew(
{
*this = (StorageReadAzure)
{
.interface = &storageReadAzureInterface,
.storage = storage,
.name = strDup(name),
.offset = offset,
.limit = varDup(limit),
.versionId = strDup(versionId),
};
}
OBJ_NEW_END();
FUNCTION_LOG_RETURN(
STORAGE_READ,
storageReadNewP(
this, STORAGE_AZURE_TYPE, name, ignoreMissing, offset, limit, &storageReadAzureInterface, .version = version,
.versionId = versionId, .retry = true));
FUNCTION_LOG_RETURN(STORAGE_READ_AZURE, this);
}

18
src/storage/azure/read.h
View File

@@ -7,11 +7,23 @@ Azure Storage Read
#include "storage/azure/storage.intern.h"
#include "storage/read.h"
/***********************************************************************************************************************************
Object type
***********************************************************************************************************************************/
typedef struct StorageReadAzure StorageReadAzure;
/***********************************************************************************************************************************
Constructors
***********************************************************************************************************************************/
FN_EXTERN StorageRead *storageReadAzureNew(
StorageAzure *storage, const String *name, bool ignoreMissing, uint64_t offset, const Variant *limit, bool version,
const String *versionId);
FN_EXTERN StorageReadAzure *storageReadAzureNew(
StorageAzure *storage, const String *name, uint64_t offset, const Variant *limit, const String *versionId);
/***********************************************************************************************************************************
Macros for function logging
***********************************************************************************************************************************/
#define FUNCTION_LOG_STORAGE_READ_AZURE_TYPE \
StorageReadAzure *
#define FUNCTION_LOG_STORAGE_READ_AZURE_FORMAT(value, buffer, bufferSize) \
objNameToLog(value, "StorageReadAzure", buffer, bufferSize)
#endif

15
src/storage/azure/storage.c
View File

@@ -686,30 +686,27 @@ storageAzureList(THIS_VOID, const String *const path, const StorageInfoLevel lev
}
/**********************************************************************************************************************************/
static StorageRead *
storageAzureNewRead(THIS_VOID, const String *const file, const bool ignoreMissing, const StorageInterfaceNewReadParam param)
static void *
storageAzureNewRead(THIS_VOID, const String *const file, const StorageInterfaceNewReadParam param)
{
THIS(StorageAzure);
FUNCTION_LOG_BEGIN(logLevelDebug);
FUNCTION_LOG_PARAM(STORAGE_AZURE, this);
FUNCTION_LOG_PARAM(STRING, file);
FUNCTION_LOG_PARAM(BOOL, ignoreMissing);
FUNCTION_LOG_PARAM(UINT64, param.offset);
FUNCTION_LOG_PARAM(VARIANT, param.limit);
FUNCTION_LOG_PARAM(BOOL, param.version);
FUNCTION_LOG_PARAM(STRING, param.versionId);
FUNCTION_LOG_END();
ASSERT(this != NULL);
ASSERT(file != NULL);
FUNCTION_LOG_RETURN(
STORAGE_READ, storageReadAzureNew(this, file, ignoreMissing, param.offset, param.limit, param.version, param.versionId));
FUNCTION_LOG_RETURN(STORAGE_READ_AZURE, storageReadAzureNew(this, file, param.offset, param.limit, param.versionId));
}
/**********************************************************************************************************************************/
static StorageWrite *
static void *
storageAzureNewWrite(THIS_VOID, const String *const file, const StorageInterfaceNewWriteParam param)
{
THIS(StorageAzure);
@@ -728,7 +725,7 @@ storageAzureNewWrite(THIS_VOID, const String *const file, const StorageInterface
ASSERT(param.group == NULL);
ASSERT(param.timeModified == 0);
FUNCTION_LOG_RETURN(STORAGE_WRITE, storageWriteAzureNew(this, file, this->fileId++, this->blockSize));
FUNCTION_LOG_RETURN(STORAGE_WRITE_AZURE, storageWriteAzureNew(this, file, this->fileId++, this->blockSize));
}
/**********************************************************************************************************************************/
@@ -834,7 +831,7 @@ storageAzureRemove(THIS_VOID, const String *const file, const StorageInterfaceRe
/**********************************************************************************************************************************/
static const StorageInterface storageInterfaceAzure =
{
.feature = 1 << storageFeatureVersioning,
.feature = 1 << storageFeatureVersioning | 1 << storageFeatureReadRetry,
.info = storageAzureInfo,
.list = storageAzureList,

30
src/storage/azure/write.c
View File

@@ -41,25 +41,18 @@ STRING_STATIC(AZURE_XML_TAG_UNCOMMITTED_STR, "Uncommitted
/***********************************************************************************************************************************
Object type
***********************************************************************************************************************************/
typedef struct StorageWriteAzure
struct StorageWriteAzure
{
StorageWriteInterface interface; // Interface
const StorageWriteInterface *interface; // Interface
StorageAzure *storage; // Storage that created this object
const String *name; // File name
HttpRequest *request; // Async block upload request
uint64_t fileId; // Id to used to make file block identifiers unique
size_t blockSize; // Size of blocks for multi-block upload
Buffer *blockBuffer; // Block buffer (stores data until blockSize is reached)
StringList *blockIdList; // List of uploaded block ids
} StorageWriteAzure;
/***********************************************************************************************************************************
Macros for function logging
***********************************************************************************************************************************/
#define FUNCTION_LOG_STORAGE_WRITE_AZURE_TYPE \
StorageWriteAzure *
#define FUNCTION_LOG_STORAGE_WRITE_AZURE_FORMAT(value, buffer, bufferSize) \
objNameToLog(value, "StorageWriteAzure", buffer, bufferSize)
};
/***********************************************************************************************************************************
Flush bytes to upload block
@@ -125,7 +118,7 @@ storageWriteAzureBlockAsync(StorageWriteAzure *const this)
MEM_CONTEXT_OBJ_BEGIN(this)
{
this->request = storageAzureRequestAsyncP(
this->storage, HTTP_VERB_PUT_STR, .path = this->interface.name, .query = query, .content = this->blockBuffer);
this->storage, HTTP_VERB_PUT_STR, .path = this->name, .query = query, .content = this->blockBuffer);
}
MEM_CONTEXT_OBJ_END();
@@ -225,7 +218,7 @@ storageWriteAzureClose(THIS_VOID)
// Finalize the multi-block upload
storageAzureRequestP(
this->storage, HTTP_VERB_PUT_STR, .path = this->interface.name,
this->storage, HTTP_VERB_PUT_STR, .path = this->name,
.query = httpQueryAdd(httpQueryNewP(), AZURE_QUERY_COMP_STR, AZURE_QUERY_VALUE_BLOCK_LIST_STR),
.content = xmlDocumentBuf(blockXml), .tag = true);
}
@@ -233,7 +226,7 @@ storageWriteAzureClose(THIS_VOID)
else
{
storageAzureRequestP(
this->storage, HTTP_VERB_PUT_STR, .path = this->interface.name,
this->storage, HTTP_VERB_PUT_STR, .path = this->name,
httpHeaderAdd(httpHeaderNew(NULL), AZURE_HEADER_BLOB_TYPE_STR, AZURE_HEADER_VALUE_BLOCK_BLOB_STR),
.content = this->blockBuffer, .tag = true);
}
@@ -248,13 +241,13 @@ storageWriteAzureClose(THIS_VOID)
}
/**********************************************************************************************************************************/
static const IoWriteInterface storageWriteAzureInterface =
static const StorageWriteInterface storageWriteAzureInterface =
{
.close = storageWriteAzureClose,
.write = storageWriteAzure,
};
FN_EXTERN StorageWrite *
FN_EXTERN StorageWriteAzure *
storageWriteAzureNew(StorageAzure *const storage, const String *const name, const uint64_t fileId, const size_t blockSize)
{
FUNCTION_LOG_BEGIN(logLevelTrace);
@@ -271,7 +264,9 @@ storageWriteAzureNew(StorageAzure *const storage, const String *const name, cons
{
*this = (StorageWriteAzure)
{
.interface = &storageWriteAzureInterface,
.storage = storage,
.name = strDup(name),
.fileId = fileId,
.blockSize = blockSize,
.blockBuffer = bufNew(0),
@@ -279,6 +274,5 @@ storageWriteAzureNew(StorageAzure *const storage, const String *const name, cons
}
OBJ_NEW_END();
FUNCTION_LOG_RETURN(
STORAGE_WRITE, storageWriteNewP(this, STORAGE_AZURE_TYPE, name, true, true, true, true, true, &storageWriteAzureInterface));
FUNCTION_LOG_RETURN(STORAGE_WRITE_AZURE, this);
}

15
src/storage/azure/write.h
View File

@@ -7,9 +7,22 @@ Azure Storage File Write
#include "storage/azure/storage.intern.h"
#include "storage/write.h"
/***********************************************************************************************************************************
Object type
***********************************************************************************************************************************/
typedef struct StorageWriteAzure StorageWriteAzure;
/***********************************************************************************************************************************
Constructors
***********************************************************************************************************************************/
FN_EXTERN StorageWrite *storageWriteAzureNew(StorageAzure *storage, const String *name, uint64_t fileId, size_t blockSize);
FN_EXTERN StorageWriteAzure *storageWriteAzureNew(StorageAzure *storage, const String *name, uint64_t fileId, size_t blockSize);
/***********************************************************************************************************************************
Macros for function logging
***********************************************************************************************************************************/
#define FUNCTION_LOG_STORAGE_WRITE_AZURE_TYPE \
StorageWriteAzure *
#define FUNCTION_LOG_STORAGE_WRITE_AZURE_FORMAT(value, buffer, bufferSize) \
objNameToLog(value, "StorageWriteAzure", buffer, bufferSize)
#endif

71
src/storage/gcs/read.c
View File

@@ -19,32 +19,30 @@ STRING_STATIC(GCS_QUERY_ALT_STR, "alt");
/***********************************************************************************************************************************
Object type
***********************************************************************************************************************************/
typedef struct StorageReadGcs
struct StorageReadGcs
{
StorageReadInterface interface; // Interface
const StorageReadInterface *interface; // Interface
StorageGcs *storage; // Storage that created this object
HttpResponse *httpResponse; // HTTP response
} StorageReadGcs;
const String *name; // File name
uint64_t offset; // Read offset
const Variant *limit; // Read limit (NULL for no limit)
const String *versionId; // Version id (NULL for most recent)
/***********************************************************************************************************************************
Macros for function logging
***********************************************************************************************************************************/
#define FUNCTION_LOG_STORAGE_READ_GCS_TYPE \
StorageReadGcs *
#define FUNCTION_LOG_STORAGE_READ_GCS_FORMAT(value, buffer, bufferSize) \
objNameToLog(value, "StorageReadGcs", buffer, bufferSize)
HttpResponse *httpResponse; // HTTP response
};
/***********************************************************************************************************************************
Open the file
***********************************************************************************************************************************/
static bool
storageReadGcsOpen(THIS_VOID)
storageReadGcsOpen(THIS_VOID, const bool ignoreMissing)
{
THIS(StorageReadGcs);
FUNCTION_LOG_BEGIN(logLevelTrace);
FUNCTION_LOG_PARAM(STORAGE_READ_GCS, this);
FUNCTION_LOG_PARAM(BOOL, ignoreMissing);
FUNCTION_LOG_END();
ASSERT(this != NULL);
@@ -52,32 +50,28 @@ storageReadGcsOpen(THIS_VOID)
bool result = false;
// Read if not versioned or if versionId is not null
if (!this->interface.version || this->interface.versionId != NULL)
MEM_CONTEXT_OBJ_BEGIN(this)
{
// Request the file
MEM_CONTEXT_OBJ_BEGIN(this)
{
HttpQuery *const query = httpQueryAdd(httpQueryNewP(), GCS_QUERY_ALT_STR, GCS_QUERY_MEDIA_STR);
HttpQuery *const query = httpQueryAdd(httpQueryNewP(), GCS_QUERY_ALT_STR, GCS_QUERY_MEDIA_STR);
if (this->interface.versionId)
httpQueryAdd(query, varStr(GCS_JSON_GENERATION_VAR), this->interface.versionId);
if (this->versionId != NULL)
httpQueryAdd(query, varStr(GCS_JSON_GENERATION_VAR), this->versionId);
this->httpResponse = storageGcsRequestP(
this->storage, HTTP_VERB_GET_STR, .object = this->interface.name,
.header = httpHeaderPutRange(httpHeaderNew(NULL), this->interface.offset, this->interface.limit),
.allowMissing = true, .contentIo = true, .query = query);
}
MEM_CONTEXT_OBJ_END();
this->httpResponse = storageGcsRequestP(
this->storage, HTTP_VERB_GET_STR, .object = this->name,
.header = httpHeaderPutRange(httpHeaderNew(NULL), this->offset, this->limit), .allowMissing = true, .contentIo = true,
.query = query);
// If file exists
if (httpResponseCodeOk(this->httpResponse))
{
result = true;
}
// Else error unless ignore missing
else if (!this->interface.ignoreMissing)
THROW_FMT(FileMissingError, STORAGE_ERROR_READ_MISSING, strZ(this->interface.name));
else if (!ignoreMissing)
THROW_FMT(FileMissingError, STORAGE_ERROR_READ_MISSING, strZ(this->name));
}
MEM_CONTEXT_OBJ_END();
FUNCTION_LOG_RETURN(BOOL, result);
}
@@ -143,7 +137,7 @@ storageReadGcsClose(THIS_VOID)
}
/**********************************************************************************************************************************/
static const IoReadInterface storageReadGcsInterface =
static const StorageReadInterface storageReadGcsInterface =
{
.close = storageReadGcsClose,
.eof = storageReadGcsEof,
@@ -151,18 +145,16 @@ static const IoReadInterface storageReadGcsInterface =
.read = storageReadGcs,
};
FN_EXTERN StorageRead *
FN_EXTERN StorageReadGcs *
storageReadGcsNew(
StorageGcs *const storage, const String *const name, const bool ignoreMissing, const uint64_t offset,
const Variant *const limit, const bool version, const String *const versionId)
StorageGcs *const storage, const String *const name, const uint64_t offset, const Variant *const limit,
const String *const versionId)
{
FUNCTION_LOG_BEGIN(logLevelTrace);
FUNCTION_LOG_PARAM(STORAGE_GCS, storage);
FUNCTION_LOG_PARAM(STRING, name);
FUNCTION_LOG_PARAM(BOOL, ignoreMissing);
FUNCTION_LOG_PARAM(UINT64, offset);
FUNCTION_LOG_PARAM(VARIANT, limit);
FUNCTION_LOG_PARAM(BOOL, version);
FUNCTION_LOG_PARAM(STRING, versionId);
FUNCTION_LOG_END();
@@ -173,14 +165,15 @@ storageReadGcsNew(
{
*this = (StorageReadGcs)
{
.interface = &storageReadGcsInterface,
.storage = storage,
.name = strDup(name),
.offset = offset,
.limit = varDup(limit),
.versionId = strDup(versionId),
};
}
OBJ_NEW_END();
FUNCTION_LOG_RETURN(
STORAGE_READ,
storageReadNewP(
this, STORAGE_GCS_TYPE, name, ignoreMissing, offset, limit, &storageReadGcsInterface, .version = version,
.versionId = versionId, .retry = true));
FUNCTION_LOG_RETURN(STORAGE_READ_GCS, this);
}

18
src/storage/gcs/read.h
View File

@@ -7,11 +7,23 @@ GCS Storage Read
#include "storage/gcs/storage.intern.h"
#include "storage/read.h"
/***********************************************************************************************************************************
Object type
***********************************************************************************************************************************/
typedef struct StorageReadGcs StorageReadGcs;
/***********************************************************************************************************************************
Constructors
***********************************************************************************************************************************/
FN_EXTERN StorageRead *storageReadGcsNew(
StorageGcs *storage, const String *name, bool ignoreMissing, uint64_t offset, const Variant *limit, bool version,
const String *versionId);
FN_EXTERN StorageReadGcs *storageReadGcsNew(
StorageGcs *storage, const String *name, uint64_t offset, const Variant *limit, const String *versionId);
/***********************************************************************************************************************************
Macros for function logging
***********************************************************************************************************************************/
#define FUNCTION_LOG_STORAGE_READ_GCS_TYPE \
StorageReadGcs *
#define FUNCTION_LOG_STORAGE_READ_GCS_FORMAT(value, buffer, bufferSize) \
objNameToLog(value, "StorageReadGcs", buffer, bufferSize)
#endif

15
src/storage/gcs/storage.c
View File

@@ -922,30 +922,27 @@ storageGcsList(THIS_VOID, const String *const path, const StorageInfoLevel level
}
/**********************************************************************************************************************************/
static StorageRead *
storageGcsNewRead(THIS_VOID, const String *const file, const bool ignoreMissing, const StorageInterfaceNewReadParam param)
static void *
storageGcsNewRead(THIS_VOID, const String *const file, const StorageInterfaceNewReadParam param)
{
THIS(StorageGcs);
FUNCTION_LOG_BEGIN(logLevelDebug);
FUNCTION_LOG_PARAM(STORAGE_GCS, this);
FUNCTION_LOG_PARAM(STRING, file);
FUNCTION_LOG_PARAM(BOOL, ignoreMissing);
FUNCTION_LOG_PARAM(UINT64, param.offset);
FUNCTION_LOG_PARAM(VARIANT, param.limit);
FUNCTION_LOG_PARAM(BOOL, param.version);
FUNCTION_LOG_PARAM(STRING, param.versionId);
FUNCTION_LOG_END();
ASSERT(this != NULL);
ASSERT(file != NULL);
FUNCTION_LOG_RETURN(
STORAGE_READ, storageReadGcsNew(this, file, ignoreMissing, param.offset, param.limit, param.version, param.versionId));
FUNCTION_LOG_RETURN(STORAGE_READ_GCS, storageReadGcsNew(this, file, param.offset, param.limit, param.versionId));
}
/**********************************************************************************************************************************/
static StorageWrite *
static void *
storageGcsNewWrite(THIS_VOID, const String *const file, const StorageInterfaceNewWriteParam param)
{
THIS(StorageGcs);
@@ -964,7 +961,7 @@ storageGcsNewWrite(THIS_VOID, const String *const file, const StorageInterfaceNe
ASSERT(param.group == NULL);
ASSERT(param.timeModified == 0);
FUNCTION_LOG_RETURN(STORAGE_WRITE, storageWriteGcsNew(this, file, this->chunkSize, this->tag != NULL));
FUNCTION_LOG_RETURN(STORAGE_WRITE_GCS, storageWriteGcsNew(this, file, this->chunkSize, this->tag != NULL));
}
/**********************************************************************************************************************************/
@@ -1173,7 +1170,7 @@ storageGcsRemove(THIS_VOID, const String *const file, const StorageInterfaceRemo
/**********************************************************************************************************************************/
static const StorageInterface storageInterfaceGcs =
{
.feature = 1 << storageFeatureVersioning,
.feature = 1 << storageFeatureVersioning | 1 << storageFeatureReadRetry,
.info = storageGcsInfo,
.list = storageGcsList,

32
src/storage/gcs/write.c
View File

@@ -29,10 +29,11 @@ STRING_STATIC(GCS_QUERY_FIELDS_VALUE_STR, GCS_JSON_MD5
/***********************************************************************************************************************************
Object type
***********************************************************************************************************************************/
typedef struct StorageWriteGcs
struct StorageWriteGcs
{
StorageWriteInterface interface; // Interface
const StorageWriteInterface *interface; // Interface
StorageGcs *storage; // Storage that created this object
const String *name; // File name
HttpRequest *request; // Async chunk upload request
size_t chunkSize; // Size of chunks for resumable upload
@@ -42,15 +43,7 @@ typedef struct StorageWriteGcs
const String *uploadId; // Id for resumable upload
uint64_t uploadTotal; // Total bytes uploaded
IoFilter *md5hash; // MD5 hash of file
} StorageWriteGcs;
/***********************************************************************************************************************************
Macros for function logging
***********************************************************************************************************************************/
#define FUNCTION_LOG_STORAGE_WRITE_GCS_TYPE \
StorageWriteGcs *
#define FUNCTION_LOG_STORAGE_WRITE_GCS_FORMAT(value, buffer, bufferSize) \
objNameToLog(value, "StorageWriteGcs", buffer, bufferSize)
};
/***********************************************************************************************************************************
Verify upload
@@ -78,7 +71,7 @@ storageWriteGcsVerify(StorageWriteGcs *const this, HttpResponse *const response)
{
THROW_FMT(
FormatError, "expected md5 '%s' for '%s' but actual is '%s'", strZ(strNewEncode(encodingHex, md5expected)),
strZ(this->interface.name), strZ(strNewEncode(encodingHex, md5actual)));
strZ(this->name), strZ(strNewEncode(encodingHex, md5actual)));
}
// Check the size when available
@@ -91,7 +84,7 @@ storageWriteGcsVerify(StorageWriteGcs *const this, HttpResponse *const response)
if (size != this->uploadTotal)
{
THROW_FMT(
FormatError, "expected size %" PRIu64 " for '%s' but actual is %" PRIu64, size, strZ(this->interface.name),
FormatError, "expected size %" PRIu64 " for '%s' but actual is %" PRIu64, size, strZ(this->name),
this->uploadTotal);
}
}
@@ -149,7 +142,7 @@ storageWriteGcsBlockAsync(StorageWriteGcs *const this, const bool done)
// Build query
HttpQuery *const query = httpQueryNewP();
httpQueryAdd(query, GCS_QUERY_NAME_STR, strSub(this->interface.name, 1));
httpQueryAdd(query, GCS_QUERY_NAME_STR, strSub(this->name, 1));
httpQueryAdd(query, GCS_QUERY_UPLOAD_TYPE_STR, GCS_QUERY_RESUMABLE_STR);
// Get the upload id
@@ -285,7 +278,7 @@ storageWriteGcsClose(THIS_VOID)
// Upload file
HttpQuery *query = httpQueryNewP();
httpQueryAdd(query, GCS_QUERY_NAME_STR, strSub(this->interface.name, 1));
httpQueryAdd(query, GCS_QUERY_NAME_STR, strSub(this->name, 1));
httpQueryAdd(query, GCS_QUERY_UPLOAD_TYPE_STR, GCS_QUERY_MEDIA_STR);
httpQueryAdd(query, GCS_QUERY_FIELDS_STR, GCS_QUERY_FIELDS_VALUE_STR);
@@ -307,13 +300,13 @@ storageWriteGcsClose(THIS_VOID)
}
/**********************************************************************************************************************************/
static const IoWriteInterface storageWriteGcsInterface =
static const StorageWriteInterface storageWriteGcsInterface =
{
.close = storageWriteGcsClose,
.write = storageWriteGcs,
};
FN_EXTERN StorageWrite *
FN_EXTERN StorageWriteGcs *
storageWriteGcsNew(StorageGcs *const storage, const String *const name, const size_t chunkSize, const bool tag)
{
FUNCTION_LOG_BEGIN(logLevelTrace);
@@ -330,7 +323,9 @@ storageWriteGcsNew(StorageGcs *const storage, const String *const name, const si
{
*this = (StorageWriteGcs)
{
.interface = &storageWriteGcsInterface,
.storage = storage,
.name = strDup(name),
.chunkSize = chunkSize,
.chunkBuffer = bufNew(0),
.md5hash = cryptoHashNew(hashTypeMd5),
@@ -339,6 +334,5 @@ storageWriteGcsNew(StorageGcs *const storage, const String *const name, const si
}
OBJ_NEW_END();
FUNCTION_LOG_RETURN(
STORAGE_WRITE, storageWriteNewP(this, STORAGE_GCS_TYPE, name, true, true, true, true, true, &storageWriteGcsInterface));
FUNCTION_LOG_RETURN(STORAGE_WRITE_GCS, this);
}

15
src/storage/gcs/write.h
View File

@@ -7,9 +7,22 @@ GCS Storage File Write
#include "storage/gcs/storage.intern.h"
#include "storage/write.h"
/***********************************************************************************************************************************
Object type
***********************************************************************************************************************************/
typedef struct StorageWriteGcs StorageWriteGcs;
/***********************************************************************************************************************************
Constructors
***********************************************************************************************************************************/
FN_EXTERN StorageWrite *storageWriteGcsNew(StorageGcs *storage, const String *name, size_t chunkSize, bool tag);
FN_EXTERN StorageWriteGcs *storageWriteGcsNew(StorageGcs *storage, const String *name, size_t chunkSize, bool tag);
/***********************************************************************************************************************************
Macros for function logging
***********************************************************************************************************************************/
#define FUNCTION_LOG_STORAGE_WRITE_GCS_TYPE \
StorageWriteGcs *
#define FUNCTION_LOG_STORAGE_WRITE_GCS_FORMAT(value, buffer, bufferSize) \
objNameToLog(value, "StorageWriteGcs", buffer, bufferSize)
#endif

58
src/storage/posix/read.c
View File

@@ -16,23 +16,19 @@ Posix Storage Read
/***********************************************************************************************************************************
Object types
***********************************************************************************************************************************/
typedef struct StorageReadPosix
struct StorageReadPosix
{
StorageReadInterface interface; // Interface
const StorageReadInterface *interface; // Interface
StoragePosix *storage; // Storage that created this object
const String *name; // File name
uint64_t offset; // Read offset
const Variant *limit; // Read limit (NULL for no limit)
int fd; // File descriptor
uint64_t current; // Current bytes read from file
bool eof;
} StorageReadPosix;
/***********************************************************************************************************************************
Macros for function logging
***********************************************************************************************************************************/
#define FUNCTION_LOG_STORAGE_READ_POSIX_TYPE \
StorageReadPosix *
#define FUNCTION_LOG_STORAGE_READ_POSIX_FORMAT(value, buffer, bufferSize) \
objNameToLog(value, "StorageReadPosix", buffer, bufferSize)
};
/***********************************************************************************************************************************
Close file descriptor
@@ -49,7 +45,7 @@ storageReadPosixFreeResource(THIS_VOID)
ASSERT(this != NULL);
if (this->fd != -1)
THROW_ON_SYS_ERROR_FMT(close(this->fd) == -1, FileCloseError, STORAGE_ERROR_READ_CLOSE, strZ(this->interface.name));
THROW_ON_SYS_ERROR_FMT(close(this->fd) == -1, FileCloseError, STORAGE_ERROR_READ_CLOSE, strZ(this->name));
FUNCTION_LOG_RETURN_VOID();
}
@@ -58,30 +54,31 @@ storageReadPosixFreeResource(THIS_VOID)
Open the file
***********************************************************************************************************************************/
static bool
storageReadPosixOpen(THIS_VOID)
storageReadPosixOpen(THIS_VOID, const bool ignoreMissing)
{
THIS(StorageReadPosix);
FUNCTION_LOG_BEGIN(logLevelTrace);
FUNCTION_LOG_PARAM(STORAGE_READ_POSIX, this);
FUNCTION_LOG_PARAM(BOOL, ignoreMissing);
FUNCTION_LOG_END();
ASSERT(this != NULL);
ASSERT(this->fd == -1);
// Open the file
this->fd = open(strZ(this->interface.name), O_RDONLY, 0);
this->fd = open(strZ(this->name), O_RDONLY, 0);
// Handle errors
if (this->fd == -1)
{
if (errno == ENOENT) // {vm_covered}
{
if (!this->interface.ignoreMissing)
THROW_FMT(FileMissingError, STORAGE_ERROR_READ_MISSING, strZ(this->interface.name));
if (!ignoreMissing)
THROW_FMT(FileMissingError, STORAGE_ERROR_READ_MISSING, strZ(this->name));
}
else
THROW_SYS_ERROR_FMT(FileOpenError, STORAGE_ERROR_READ_OPEN, strZ(this->interface.name)); // {vm_covered}
THROW_SYS_ERROR_FMT(FileOpenError, STORAGE_ERROR_READ_OPEN, strZ(this->name)); // {vm_covered}
}
// Else success
else
@@ -90,11 +87,11 @@ storageReadPosixOpen(THIS_VOID)
memContextCallbackSet(objMemContext(this), storageReadPosixFreeResource, this);
// Seek to offset
if (this->interface.offset != 0)
if (this->offset != 0)
{
THROW_ON_SYS_ERROR_FMT(
lseek(this->fd, (off_t)this->interface.offset, SEEK_SET) == -1, FileOpenError, STORAGE_ERROR_READ_SEEK,
this->interface.offset, strZ(this->interface.name));
lseek(this->fd, (off_t)this->offset, SEEK_SET) == -1, FileOpenError, STORAGE_ERROR_READ_SEEK, this->offset,
strZ(this->name));
}
}
@@ -125,7 +122,7 @@ storageReadPosix(THIS_VOID, Buffer *const buffer, const bool block)
{
// Rather than enable/disable limit checking just use a big number when there is no limit. We can feel pretty confident that
// no files will be > UINT64_MAX in size.
const uint64_t limit = this->interface.limit == NULL ? UINT64_MAX : varUInt64(this->interface.limit);
const uint64_t limit = this->limit == NULL ? UINT64_MAX : varUInt64(this->limit);
// Determine expected bytes to read. If remaining size in the buffer would exceed the limit then reduce the expected read.
size_t expectedBytes = bufRemains(buffer);
@@ -138,7 +135,7 @@ storageReadPosix(THIS_VOID, Buffer *const buffer, const bool block)
// Error occurred during read
if (actualBytes == -1)
THROW_SYS_ERROR_FMT(FileReadError, "unable to read '%s'", strZ(this->interface.name));
THROW_SYS_ERROR_FMT(FileReadError, "unable to read '%s'", strZ(this->name));
// Update amount of buffer used
bufUsedInc(buffer, (size_t)actualBytes);
@@ -209,7 +206,7 @@ storageReadPosixFd(const THIS_VOID)
}
/**********************************************************************************************************************************/
static const IoReadInterface storageReadPosixInterface =
static const StorageReadInterface storageReadPosixInterface =
{
.close = storageReadPosixClose,
.eof = storageReadPosixEof,
@@ -218,14 +215,12 @@ static const IoReadInterface storageReadPosixInterface =
.read = storageReadPosix,
};
FN_EXTERN StorageRead *
storageReadPosixNew(
StoragePosix *const storage, const String *const name, const bool ignoreMissing, const uint64_t offset,
const Variant *const limit)
FN_EXTERN StorageReadPosix *
storageReadPosixNew(StoragePosix *const storage, const String *const name, const uint64_t offset, const Variant *const limit)
{
FUNCTION_LOG_BEGIN(logLevelTrace);
FUNCTION_LOG_PARAM(STORAGE_POSIX, storage);
FUNCTION_LOG_PARAM(STRING, name);
FUNCTION_LOG_PARAM(BOOL, ignoreMissing);
FUNCTION_LOG_PARAM(UINT64, offset);
FUNCTION_LOG_PARAM(VARIANT, limit);
FUNCTION_LOG_END();
@@ -236,12 +231,15 @@ storageReadPosixNew(
{
*this = (StorageReadPosix)
{
.interface = &storageReadPosixInterface,
.storage = storage,
.name = strDup(name),
.offset = offset,
.limit = varDup(limit),
.fd = -1,
};
}
OBJ_NEW_END();
FUNCTION_LOG_RETURN(
STORAGE_READ, storageReadNewP(this, STORAGE_POSIX_TYPE, name, ignoreMissing, offset, limit, &storageReadPosixInterface));
FUNCTION_LOG_RETURN(STORAGE_READ_POSIX, this);
}

16
src/storage/posix/read.h
View File

@@ -7,10 +7,22 @@ Posix Storage Read
#include "storage/posix/storage.intern.h"
#include "storage/read.h"
/***********************************************************************************************************************************
Object type
***********************************************************************************************************************************/
typedef struct StorageReadPosix StorageReadPosix;
/***********************************************************************************************************************************
Constructors
***********************************************************************************************************************************/
FN_EXTERN StorageRead *storageReadPosixNew(
StoragePosix *storage, const String *name, bool ignoreMissing, uint64_t offset, const Variant *limit);
FN_EXTERN StorageReadPosix *storageReadPosixNew(StoragePosix *storage, const String *name, uint64_t offset, const Variant *limit);
/***********************************************************************************************************************************
Macros for function logging
***********************************************************************************************************************************/
#define FUNCTION_LOG_STORAGE_READ_POSIX_TYPE \
StorageReadPosix *
#define FUNCTION_LOG_STORAGE_READ_POSIX_FORMAT(value, buffer, bufferSize) \
objNameToLog(value, "StorageReadPosix", buffer, bufferSize)
#endif

12
src/storage/posix/storage.c
View File

@@ -342,29 +342,27 @@ storagePosixMove(THIS_VOID, StorageRead *const source, StorageWrite *const desti
}
/**********************************************************************************************************************************/
static StorageRead *
storagePosixNewRead(THIS_VOID, const String *const file, const bool ignoreMissing, const StorageInterfaceNewReadParam param)
static void *
storagePosixNewRead(THIS_VOID, const String *const file, const StorageInterfaceNewReadParam param)
{
THIS(StoragePosix);
FUNCTION_LOG_BEGIN(logLevelTrace);
FUNCTION_LOG_PARAM(STORAGE_POSIX, this);
FUNCTION_LOG_PARAM(STRING, file);
FUNCTION_LOG_PARAM(BOOL, ignoreMissing);
FUNCTION_LOG_PARAM(UINT64, param.offset);
FUNCTION_LOG_PARAM(VARIANT, param.limit);
FUNCTION_LOG_END();
ASSERT(this != NULL);
ASSERT(file != NULL);
ASSERT(!param.version);
ASSERT(param.versionId == NULL);
FUNCTION_LOG_RETURN(STORAGE_READ, storageReadPosixNew(this, file, ignoreMissing, param.offset, param.limit));
FUNCTION_LOG_RETURN(STORAGE_READ_POSIX, storageReadPosixNew(this, file, param.offset, param.limit));
}
/**********************************************************************************************************************************/
static StorageWrite *
static void *
storagePosixNewWrite(THIS_VOID, const String *const file, const StorageInterfaceNewWriteParam param)
{
THIS(StoragePosix);
@@ -388,7 +386,7 @@ storagePosixNewWrite(THIS_VOID, const String *const file, const StorageInterface
ASSERT(file != NULL);
FUNCTION_LOG_RETURN(
STORAGE_WRITE,
STORAGE_WRITE_POSIX,
storageWritePosixNew(
this, file, param.modeFile, param.modePath, param.user, param.group, param.timeModified, param.createPath,
param.syncFile, this->interface.pathSync != NULL ? param.syncPath : false, param.atomic, param.truncate));

94
src/storage/posix/write.c
View File

@@ -19,23 +19,27 @@ Posix Storage File write
/***********************************************************************************************************************************
Object type
***********************************************************************************************************************************/
typedef struct StorageWritePosix
struct StorageWritePosix
{
StorageWriteInterface interface; // Interface
StoragePosix *storage; // Storage that created this object
const StorageWriteInterface *interface; // Interface
StoragePosix *storage; // Storage that created this object
const String *name; // File name
mode_t modeFile; // File mode
mode_t modePath; // Path mode
bool createPath; // Create path if missing
bool atomic; // Atomic write
bool truncate; // Truncate existing file
bool syncPath; // Sync path after write
bool syncFile; // Sync file after write
const String *user; // User owner
const String *group; // Group owner
time_t timeModified; // Modified time
const String *nameTmp;
const String *path;
int fd; // File descriptor
} StorageWritePosix;
/***********************************************************************************************************************************
Macros for function logging
***********************************************************************************************************************************/
#define FUNCTION_LOG_STORAGE_WRITE_POSIX_TYPE \
StorageWritePosix *
#define FUNCTION_LOG_STORAGE_WRITE_POSIX_FORMAT(value, buffer, bufferSize) \
objNameToLog(value, "StorageWritePosix", buffer, bufferSize)
int fd; // File descriptor
};
/***********************************************************************************************************************************
File open constants
@@ -91,42 +95,42 @@ storageWritePosixOpen(THIS_VOID)
ASSERT(this->fd == -1);
// Open the file
const int flags = O_CREAT | O_WRONLY | (this->interface.truncate ? O_TRUNC : 0);
const int flags = O_CREAT | O_WRONLY | (this->truncate ? O_TRUNC : 0);
this->fd = open(strZ(this->nameTmp), flags, this->interface.modeFile);
this->fd = open(strZ(this->nameTmp), flags, this->modeFile);
// Attempt to create the path if it is missing
if (this->fd == -1 && errno == ENOENT && this->interface.createPath) // {vm_covered}
if (this->fd == -1 && errno == ENOENT && this->createPath) // {vm_covered}
{
// Create the path
storageInterfacePathCreateP(this->storage, this->path, false, false, this->interface.modePath);
storageInterfacePathCreateP(this->storage, this->path, false, false, this->modePath);
// Open file again
this->fd = open(strZ(this->nameTmp), flags, this->interface.modeFile);
this->fd = open(strZ(this->nameTmp), flags, this->modeFile);
}
// Handle errors
if (this->fd == -1)
{
if (errno == ENOENT) // {vm_covered}
THROW_FMT(FileMissingError, STORAGE_ERROR_WRITE_MISSING, strZ(this->interface.name));
THROW_FMT(FileMissingError, STORAGE_ERROR_WRITE_MISSING, strZ(this->name));
else
THROW_SYS_ERROR_FMT(FileOpenError, STORAGE_ERROR_WRITE_OPEN, strZ(this->interface.name)); // {vm_covered}
THROW_SYS_ERROR_FMT(FileOpenError, STORAGE_ERROR_WRITE_OPEN, strZ(this->name)); // {vm_covered}
}
// Set free callback to ensure the file descriptor is freed
memContextCallbackSet(objMemContext(this), storageWritePosixFreeResource, this);
// Update user/group owner
if (this->interface.user != NULL || this->interface.group != NULL)
if (this->user != NULL || this->group != NULL)
{
MEM_CONTEXT_TEMP_BEGIN()
{
const StorageInfo info = storageInterfaceInfoP(
this->storage, this->nameTmp, storageInfoLevelDetail, .followLink = true);
ASSERT(info.exists);
uid_t updateUserId = userIdFromName(this->interface.user);
gid_t updateGroupId = groupIdFromName(this->interface.group);
uid_t updateUserId = userIdFromName(this->user);
gid_t updateGroupId = groupIdFromName(this->group);
if (updateUserId == (uid_t)-1)
updateUserId = info.userId;
@@ -140,10 +144,10 @@ storageWritePosixOpen(THIS_VOID)
THROW_ON_SYS_ERROR_FMT(
chown(strZ(this->nameTmp), updateUserId, updateGroupId) == -1, FileOwnerError,
"unable to set ownership for '%s' to %s%s:%s%s from %s[%u]:%s[%u]", strZ(this->nameTmp),
storageWritePosixOpenOwner(this->interface.user, "[none]"),
storageWritePosixOpenOwnerId(this->interface.user, updateUserId),
storageWritePosixOpenOwner(this->interface.group, "[none]"),
storageWritePosixOpenOwnerId(this->interface.group, updateGroupId),
storageWritePosixOpenOwner(this->user, "[none]"),
storageWritePosixOpenOwnerId(this->user, updateUserId),
storageWritePosixOpenOwner(this->group, "[none]"),
storageWritePosixOpenOwnerId(this->group, updateGroupId),
storageWritePosixOpenOwner(info.user, "[unknown]"), info.userId,
storageWritePosixOpenOwner(info.group, "[unknown]"), info.groupId);
}
@@ -218,7 +222,7 @@ storageWritePosixClose(THIS_VOID)
if (this->fd != -1)
{
// Sync the file
if (this->interface.syncFile)
if (this->syncFile)
THROW_ON_SYS_ERROR_FMT(fsync(this->fd) == -1, FileSyncError, STORAGE_ERROR_WRITE_SYNC, strZ(this->nameTmp));
// Close the file
@@ -227,24 +231,24 @@ storageWritePosixClose(THIS_VOID)
this->fd = -1;
// Update modified time
if (this->interface.timeModified != 0)
if (this->timeModified != 0)
{
THROW_ON_SYS_ERROR_FMT(
utime(
strZ(this->nameTmp),
&((struct utimbuf){.actime = this->interface.timeModified, .modtime = this->interface.timeModified})) == -1,
&((struct utimbuf){.actime = this->timeModified, .modtime = this->timeModified})) == -1,
FileInfoError, "unable to set time for '%s'", strZ(this->nameTmp));
}
// Rename from temp file
if (this->interface.atomic)
if (this->atomic)
{
if (rename(strZ(this->nameTmp), strZ(this->interface.name)) == -1)
THROW_SYS_ERROR_FMT(FileMoveError, "unable to move '%s' to '%s'", strZ(this->nameTmp), strZ(this->interface.name));
if (rename(strZ(this->nameTmp), strZ(this->name)) == -1)
THROW_SYS_ERROR_FMT(FileMoveError, "unable to move '%s' to '%s'", strZ(this->nameTmp), strZ(this->name));
}
// Sync the path
if (this->interface.syncPath)
if (this->syncPath)
storageInterfacePathSyncP(this->storage, this->path);
}
@@ -269,7 +273,7 @@ storageWritePosixFd(const THIS_VOID)
}
/**********************************************************************************************************************************/
static const IoWriteInterface storageWritePosixInterface =
static const StorageWriteInterface storageWritePosixInterface =
{
.close = storageWritePosixClose,
.fd = storageWritePosixFd,
@@ -278,7 +282,7 @@ static const IoWriteInterface storageWritePosixInterface =
.write = storageWritePosix,
};
FN_EXTERN StorageWrite *
FN_EXTERN StorageWritePosix *
storageWritePosixNew(
StoragePosix *const storage, const String *const name, const mode_t modeFile, const mode_t modePath, const String *const user,
const String *const group, const time_t timeModified, const bool createPath, const bool syncFile, const bool syncPath,
@@ -308,7 +312,19 @@ storageWritePosixNew(
{
*this = (StorageWritePosix)
{
.interface = &storageWritePosixInterface,
.storage = storage,
.name = strDup(name),
.modeFile = modeFile,
.modePath = modePath,
.createPath = createPath,
.atomic = atomic,
.truncate = truncate,
.syncFile = syncFile,
.syncPath = syncPath,
.user = strDup(user),
.group = strDup(group),
.timeModified = timeModified,
.nameTmp = atomic ? strNewFmt("%s." STORAGE_FILE_TEMP_EXT, strZ(name)) : strDup(name),
.path = strPath(name),
.fd = -1,
@@ -316,9 +332,5 @@ storageWritePosixNew(
}
OBJ_NEW_END();
FUNCTION_LOG_RETURN(
STORAGE_WRITE,
storageWriteNewP(
this, STORAGE_POSIX_TYPE, name, createPath, atomic, truncate, syncPath, syncFile, &storageWritePosixInterface,
.user = user, .group = group, .modePath = modePath, .modeFile = modeFile, .timeModified = timeModified));
FUNCTION_LOG_RETURN(STORAGE_WRITE_POSIX, this);
}

15
src/storage/posix/write.h
View File

@@ -7,11 +7,24 @@ Posix Storage File write
#include "storage/posix/storage.intern.h"
#include "storage/write.h"
/***********************************************************************************************************************************
Object type
***********************************************************************************************************************************/
typedef struct StorageWritePosix StorageWritePosix;
/***********************************************************************************************************************************
Constructors
***********************************************************************************************************************************/
FN_EXTERN StorageWrite *storageWritePosixNew(
FN_EXTERN StorageWritePosix *storageWritePosixNew(
StoragePosix *storage, const String *name, mode_t modeFile, mode_t modePath, const String *user, const String *group,
time_t timeModified, bool createPath, bool syncFile, bool syncPath, bool atomic, bool truncate);
/***********************************************************************************************************************************
Macros for function logging
***********************************************************************************************************************************/
#define FUNCTION_LOG_STORAGE_WRITE_POSIX_TYPE \
StorageWritePosix *
#define FUNCTION_LOG_STORAGE_WRITE_POSIX_FORMAT(value, buffer, bufferSize) \
objNameToLog(value, "StorageWritePosix", buffer, bufferSize)
#endif

133
src/storage/read.c
View File

@@ -15,10 +15,11 @@ Object type
struct StorageRead
{
StorageReadPub pub; // Publicly accessible variables
const Storage *storage; // Storage
void *driver; // Driver
IoReadInterface ioInterface; // Driver Io interface
uint64_t bytesRead; // Bytes that have been successfully read
bool retry; // Are read retries allowed?
bool compressible; // Is the read compressible?
};
/***********************************************************************************************************************************
@@ -29,15 +30,6 @@ Macros for function logging
#define FUNCTION_LOG_STORAGE_READ_INTERFACE_FORMAT(value, buffer, bufferSize) \
objNameToLog(&value, "StorageReadInterface", buffer, bufferSize)
/***********************************************************************************************************************************
Get driver interface
***********************************************************************************************************************************/
FN_INLINE_ALWAYS StorageReadInterface *
storageReadDriverInterface(StorageRead *const this)
{
return (StorageReadInterface *)this->driver;
}
/***********************************************************************************************************************************
Open the file
***********************************************************************************************************************************/
@@ -52,10 +44,11 @@ storageReadOpen(THIS_VOID)
ASSERT(this != NULL);
const bool result = this->ioInterface.open(this->driver);
bool result = false;
// Now that the file is open disable ignore missing. On retry the file must not be missing so we want a hard error.
storageReadDriverInterface(this)->ignoreMissing = false;
// Open if not versioned or if versionId is not null
if (!this->pub.version || this->pub.versionId != NULL)
result = storageReadDriverInterface(this->driver)->open(this->driver, this->pub.ignoreMissing);
FUNCTION_LOG_RETURN(BOOL, result);
}
@@ -90,7 +83,7 @@ storageRead(THIS_VOID, Buffer *const buffer, const bool block)
{
TRY_BEGIN()
{
this->ioInterface.read(this->driver, buffer, block);
storageReadDriverInterface(this->driver)->read(this->driver, buffer, block);
// Account for bytes that have been read
result += bufUsed(buffer) - bufUsedBegin;
@@ -106,28 +99,24 @@ storageRead(THIS_VOID, Buffer *const buffer, const bool block)
if (try > 1)
{
// Close the file
this->ioInterface.close(this->driver);
storageReadDriverInterface(this->driver)->close(this->driver);
// Ignore partial reads and restart from the last successful read
bufUsedSet(buffer, bufUsedBegin);
// Update offset and limit (when present) based on how many bytes have been successfully read
storageReadDriverInterface(this)->offset = storageReadInterface(this)->offset + this->bytesRead;
if (storageReadInterface(this)->limit != NULL)
// Driver with new offset/limit
MEM_CONTEXT_OBJ_BEGIN(this)
{
varFree(storageReadDriverInterface(this)->limit);
objFree(this->driver);
MEM_CONTEXT_OBJ_BEGIN(this->driver)
{
storageReadDriverInterface(this)->limit = varNewUInt64(
varUInt64(storageReadInterface(this)->limit) - this->bytesRead);
}
MEM_CONTEXT_OBJ_END();
this->driver = storageInterfaceNewReadP(
storageDriver(this->storage), this->pub.name, .compressible = this->compressible,
.offset = this->pub.offset + this->bytesRead,
.limit = this->pub.limit != NULL ? varNewUInt64(varUInt64(this->pub.limit) - this->bytesRead) : NULL,
.versionId = this->pub.versionId);
storageReadDriverInterface(this->driver)->open(this->driver, false);
}
// Open file with new offset/limit
this->ioInterface.open(this->driver);
MEM_CONTEXT_OBJ_END();
}
else
RETHROW();
@@ -156,7 +145,7 @@ storageReadClose(THIS_VOID)
ASSERT(this != NULL);
this->ioInterface.close(this->driver);
storageReadDriverInterface(this->driver)->close(this->driver);
FUNCTION_LOG_RETURN_VOID();
}
@@ -175,7 +164,7 @@ storageReadEof(THIS_VOID)
ASSERT(this != NULL);
FUNCTION_TEST_RETURN(BOOL, this->ioInterface.eof(this->driver));
FUNCTION_TEST_RETURN(BOOL, storageReadDriverInterface(this->driver)->eof(this->driver));
}
/***********************************************************************************************************************************
@@ -192,7 +181,7 @@ storageReadFd(const THIS_VOID)
ASSERT(this != NULL);
FUNCTION_TEST_RETURN(INT, this->ioInterface.fd(this->driver));
FUNCTION_TEST_RETURN(INT, storageReadDriverInterface(this->driver)->fd(this->driver));
}
/**********************************************************************************************************************************/
@@ -207,72 +196,66 @@ static const IoReadInterface storageIoReadInterface =
FN_EXTERN StorageRead *
storageReadNew(
void *const driver, const StringId type, const String *const name, const bool ignoreMissing, const uint64_t offset,
const Variant *const limit, const IoReadInterface *const ioInterface, const StorageReadNewParam param)
const Storage *const storage, const String *const name, const bool ignoreMissing, const bool compressible,
const uint64_t offset, const Variant *const limit, const bool version, const String *const versionId)
{
FUNCTION_LOG_BEGIN(logLevelTrace);
FUNCTION_LOG_PARAM_P(VOID, driver);
FUNCTION_LOG_PARAM(STRING_ID, type);
FUNCTION_LOG_PARAM(STORAGE, storage);
FUNCTION_LOG_PARAM(STRING, name);
FUNCTION_LOG_PARAM(BOOL, ignoreMissing);
FUNCTION_LOG_PARAM(BOOL, compressible);
FUNCTION_LOG_PARAM(UINT64, offset);
FUNCTION_LOG_PARAM(VARIANT, limit);
FUNCTION_LOG_PARAM_P(VOID, ioInterface);
FUNCTION_LOG_PARAM(BOOL, param.retry);
FUNCTION_LOG_PARAM(BOOL, param.version);
FUNCTION_LOG_PARAM(STRING, param.versionId);
FUNCTION_LOG_PARAM(BOOL, version);
FUNCTION_LOG_PARAM(STRING, versionId);
FUNCTION_LOG_END();
FUNCTION_AUDIT_HELPER();
ASSERT(driver != NULL);
ASSERT(type != 0);
ASSERT(storage != NULL);
ASSERT(name != NULL);
ASSERT(ioInterface != NULL);
ASSERT(ioInterface->eof != NULL);
ASSERT(ioInterface->close != NULL);
ASSERT(ioInterface->open != NULL);
ASSERT(ioInterface->read != NULL);
// Remove fd method if it does not exist in the driver
IoReadInterface storageIoReadInterfaceCopy = storageIoReadInterface;
if (ioInterface->fd == NULL)
storageIoReadInterfaceCopy.fd = NULL;
OBJ_NEW_BEGIN(StorageRead, .childQty = MEM_CONTEXT_QTY_MAX)
{
*this = (StorageRead)
{
.driver = objMove(driver, memContextCurrent()),
.ioInterface = *ioInterface,
.retry = param.retry,
.storage = storage,
.driver = storageInterfaceNewReadP(
storageDriver(storage), name, .compressible = compressible, .offset = offset, .limit = limit,
.versionId = versionId),
.retry = storageFeature(storage, storageFeatureReadRetry),
.compressible = compressible,
.pub =
{
.interface =
{
.name = strDup(name),
.ignoreMissing = ignoreMissing,
.offset = offset,
.limit = varDup(limit),
.version = param.version,
.versionId = strDup(param.versionId),
},
.type = type,
.io = ioReadNew(this, storageIoReadInterfaceCopy),
.type = storageType(storage),
.name = strDup(name),
.ignoreMissing = ignoreMissing,
.offset = offset,
.limit = varDup(limit),
.version = version,
.versionId = strDup(versionId),
},
};
ASSERT(storageReadDriverInterface(this->driver)->eof != NULL);
ASSERT(storageReadDriverInterface(this->driver)->close != NULL);
ASSERT(storageReadDriverInterface(this->driver)->open != NULL);
ASSERT(storageReadDriverInterface(this->driver)->read != NULL);
// Remove fd method if it does not exist in the driver
IoReadInterface storageIoReadInterfaceCopy = storageIoReadInterface;
if (storageReadDriverInterface(this->driver)->fd == NULL)
storageIoReadInterfaceCopy.fd = NULL;
this->pub.io = ioReadNew(this, storageIoReadInterfaceCopy);
// Set filter group when interface function exists
if (storageReadDriverInterface(this->driver)->filterGroup != NULL)
storageReadDriverInterface(this->driver)->filterGroup(this->driver, ioReadFilterGroup(storageReadIo(this)));
}
OBJ_NEW_END();
// Copy the interface into the driver and duplicate variables that must be local to the driver
MEM_CONTEXT_OBJ_BEGIN(this->driver)
{
*(storageReadDriverInterface(this)) = this->pub.interface;
storageReadDriverInterface(this)->limit = varDup(storageReadInterface(this)->limit);
}
MEM_CONTEXT_OBJ_END();
FUNCTION_LOG_RETURN(STORAGE_READ, this);
}

20
src/storage/read.h
View File

@@ -26,11 +26,23 @@ storageReadMove(StorageRead *const this, MemContext *const parentNew)
/***********************************************************************************************************************************
Getters/Setters
***********************************************************************************************************************************/
typedef struct StorageReadPub
{
StringId type; // Storage type
const String *name; // File name
uint64_t offset; // Where to start reading in the file
Variant *limit; // Limit how many bytes are read (NULL for no limit)
bool ignoreMissing; // Ignore missing file?
bool version; // Read version
const String *versionId; // File version to read
IoRead *io; // Read interface
} StorageReadPub;
// Should a missing file be ignored?
FN_INLINE_ALWAYS bool
storageReadIgnoreMissing(const StorageRead *const this)
{
return storageReadInterface(this)->ignoreMissing;
return THIS_PUB(StorageRead)->ignoreMissing;
}
// Read interface
@@ -44,21 +56,21 @@ storageReadIo(const StorageRead *const this)
FN_INLINE_ALWAYS const Variant *
storageReadLimit(const StorageRead *const this)
{
return storageReadInterface(this)->limit;
return THIS_PUB(StorageRead)->limit;
}
// File name
FN_INLINE_ALWAYS const String *
storageReadName(const StorageRead *const this)
{
return storageReadInterface(this)->name;
return THIS_PUB(StorageRead)->name;
}
// Is there a read limit? NULL for no limit.
FN_INLINE_ALWAYS uint64_t
storageReadOffset(const StorageRead *const this)
{
return storageReadInterface(this)->offset;
return THIS_PUB(StorageRead)->offset;
}
// Get file type

54
src/storage/read.intern.h
View File

@@ -5,51 +5,35 @@ Storage Read Interface Internal
#define STORAGE_READ_INTERN_H
#include "common/io/read.h"
#include "storage/storage.h"
/***********************************************************************************************************************************
Interface
***********************************************************************************************************************************/
typedef struct StorageReadInterface
{
void (*filterGroup)(void *driver, IoFilterGroup *filterGroup); // Set filter group (optional)
bool (*open)(void *driver, bool ignoreMissing); // Open read
size_t (*read)(void *driver, Buffer *buffer, bool block); // Read bytes
bool (*eof)(void *driver); // Is read eof?
int (*fd)(const void *driver); // Read file descriptor (optional)
void (*close)(void *driver); // Close read
} StorageReadInterface;
/***********************************************************************************************************************************
Constructors
***********************************************************************************************************************************/
typedef struct StorageReadInterface
{
const String *name; // File name
uint64_t offset; // Where to start reading in the file
Variant *limit; // Limit how many bytes are read (NULL for no limit)
bool ignoreMissing; // Ignore missing file?
bool version; // Read version
const String *versionId; // File version to read
} StorageReadInterface;
typedef struct StorageReadNewParam
{
VAR_PARAM_HEADER;
bool retry; // Are read retries allowed?
bool version; // Read version
const String *versionId; // File version to read
} StorageReadNewParam;
#define storageReadNewP(driver, type, name, ignoreMissing, offset, limit, ioInterface, ...) \
storageReadNew( \
driver, type, name, ignoreMissing, offset, limit, ioInterface, (StorageReadNewParam){VAR_PARAM_INIT, __VA_ARGS__})
FN_EXTERN StorageRead *storageReadNew(
void *driver, StringId type, const String *name, bool ignoreMissing, uint64_t offset, const Variant *limit,
const IoReadInterface *ioInterface, StorageReadNewParam param);
const Storage *storage, const String *name, bool ignoreMissing, bool compressible, uint64_t offset, const Variant *limit,
bool version, const String *versionId);
/***********************************************************************************************************************************
Getters/Setters
Getters
***********************************************************************************************************************************/
typedef struct StorageReadPub
{
StorageReadInterface interface; // File data (name, driver type, etc.)
IoRead *io; // Read interface
StringId type; // Storage type
} StorageReadPub;
// Read interface
FN_INLINE_ALWAYS const StorageReadInterface *
storageReadInterface(const StorageRead *const this)
storageReadDriverInterface(const void *const driver)
{
return &THIS_PUB(StorageRead)->interface;
return *(const StorageReadInterface *const *)driver;
}
#endif

16
src/storage/remote/protocol.c
View File

@@ -14,6 +14,7 @@ Remote Storage Protocol Handler
#include "storage/helper.h"
#include "storage/remote/protocol.h"
#include "storage/storage.intern.h"
#include "storage/write.intern.h"
/***********************************************************************************************************************************
Local variables
@@ -21,6 +22,7 @@ Local variables
static struct
{
MemContext *memContext; // Mem context
const Storage *storage; // Storage
void *driver; // Storage driver used for requests
const List *filterHandler; // Filter handler list
@@ -150,6 +152,7 @@ storageRemoteFeatureProtocol(PackRead *const param)
MEM_CONTEXT_NEW_BEGIN(StorageRemoteProtocol, .childQty = MEM_CONTEXT_QTY_MAX)
{
storageRemoteProtocolLocal.memContext = memContextCurrent();
storageRemoteProtocolLocal.storage = storage;
storageRemoteProtocolLocal.driver = storageDriver(storage);
}
MEM_CONTEXT_NEW_END();
@@ -425,14 +428,12 @@ storageRemoteReadOpenProtocol(PackRead *const param)
const bool ignoreMissing = pckReadBoolP(param);
const uint64_t offset = pckReadU64P(param);
const Variant *const limit = pckReadNullP(param) ? NULL : VARUINT64(pckReadU64P(param));
const bool version = pckReadBoolP(param);
const String *const versionId = pckReadStrP(param);
const Pack *const filter = pckReadPackP(param);
// Create the read object
StorageRead *const fileRead = storageInterfaceNewReadP(
storageRemoteProtocolLocal.driver, file, ignoreMissing, .offset = offset, .limit = limit, .version = version,
.versionId = versionId);
StorageRead *const fileRead = storageReadNew(
storageRemoteProtocolLocal.storage, file, ignoreMissing, false, offset, limit, versionId != NULL, versionId);
// Set filter group based on passed filters
storageRemoteFilterGroup(ioReadFilterGroup(storageReadIo(fileRead)), filter);
@@ -508,10 +509,9 @@ storageRemoteWriteOpenProtocol(PackRead *const param)
const bool atomic = pckReadBoolP(param);
const Pack *const filter = pckReadPackP(param);
StorageWrite *const fileWrite = storageInterfaceNewWriteP(
storageRemoteProtocolLocal.driver, file, .modeFile = modeFile, .modePath = modePath, .user = user, .group = group,
.timeModified = timeModified, .createPath = createPath, .syncFile = syncFile, .syncPath = syncPath, .atomic = atomic,
.truncate = true);
StorageWrite *const fileWrite = storageWriteNew(
storageRemoteProtocolLocal.storage, file, modeFile, modePath, user, group, timeModified, createPath, syncFile,
syncPath, atomic, true, false);
// Set filter group based on passed filters
storageRemoteFilterGroup(ioWriteFilterGroup(storageWriteIo(fileWrite)), filter);

93
src/storage/remote/read.c
View File

@@ -20,11 +20,16 @@ Remote Storage Read
/***********************************************************************************************************************************
Object type
***********************************************************************************************************************************/
typedef struct StorageReadRemote
struct StorageReadRemote
{
StorageReadInterface interface; // Interface
const StorageReadInterface *interface; // Interface
StorageRemote *storage; // Storage that created this object
StorageRead *read; // Storage read interface
IoFilterGroup *filterGroup; // Filter group
const String *name; // File name
uint64_t offset; // Read offset
const Variant *limit; // Read limit (NULL for no limit)
const String *versionId; // Version id (NULL for most recent)
ProtocolClient *client; // Protocol client for requests
ProtocolClientSession *session; // Protocol session for requests
@@ -38,15 +43,7 @@ typedef struct StorageReadRemote
#ifdef DEBUG
uint64_t protocolReadBytes; // How many bytes were read from the protocol layer?
#endif
} StorageReadRemote;
/***********************************************************************************************************************************
Macros for function logging
***********************************************************************************************************************************/
#define FUNCTION_LOG_STORAGE_READ_REMOTE_TYPE \
StorageReadRemote *
#define FUNCTION_LOG_STORAGE_READ_REMOTE_FORMAT(value, buffer, bufferSize) \
objNameToLog(value, "StorageReadRemote", buffer, bufferSize)
};
/***********************************************************************************************************************************
Read from a file
@@ -78,7 +75,7 @@ storageReadRemoteInternal(StorageReadRemote *const this, PackRead *const packRea
// If eof then get results
if (protocolClientSessionClosed(this->session))
{
ioFilterGroupResultAllSet(ioReadFilterGroup(storageReadIo(this->read)), pckReadPackP(packRead));
ioFilterGroupResultAllSet(this->filterGroup, pckReadPackP(packRead));
this->eofFound = true;
@@ -175,16 +172,35 @@ storageReadRemoteEof(THIS_VOID)
FUNCTION_TEST_RETURN(BOOL, this->eof);
}
/***********************************************************************************************************************************
Set filter group
***********************************************************************************************************************************/
static void
storageReadRemoteFilterGroup(THIS_VOID, IoFilterGroup *const filterGroup)
{
THIS(StorageReadRemote);
FUNCTION_TEST_BEGIN();
FUNCTION_TEST_PARAM(STORAGE_READ_REMOTE, this);
FUNCTION_TEST_PARAM(IO_FILTER_GROUP, filterGroup);
FUNCTION_TEST_END();
this->filterGroup = filterGroup;
FUNCTION_TEST_RETURN_VOID();
}
/***********************************************************************************************************************************
Open the file
***********************************************************************************************************************************/
static bool
storageReadRemoteOpen(THIS_VOID)
storageReadRemoteOpen(THIS_VOID, const bool ignoreMissing)
{
THIS(StorageReadRemote);
FUNCTION_LOG_BEGIN(logLevelTrace);
FUNCTION_LOG_PARAM(STORAGE_READ_REMOTE, this);
FUNCTION_LOG_PARAM(BOOL, ignoreMissing);
FUNCTION_LOG_END();
ASSERT(this != NULL);
@@ -195,26 +211,21 @@ storageReadRemoteOpen(THIS_VOID)
{
// If the file is compressible add compression filter on the remote
if (this->compressible)
{
ioFilterGroupAdd(
ioReadFilterGroup(storageReadIo(this->read)),
compressFilterP(compressTypeLz4, (int)this->compressLevel, .raw = true));
}
ioFilterGroupAdd(this->filterGroup, compressFilterP(compressTypeLz4, (int)this->compressLevel, .raw = true));
PackWrite *const param = protocolPackNew();
pckWriteStrP(param, this->interface.name);
pckWriteBoolP(param, this->interface.ignoreMissing);
pckWriteU64P(param, this->interface.offset);
pckWriteStrP(param, this->name);
pckWriteBoolP(param, ignoreMissing);
pckWriteU64P(param, this->offset);
if (this->interface.limit == NULL)
if (this->limit == NULL)
pckWriteNullP(param);
else
pckWriteU64P(param, varUInt64(this->interface.limit));
pckWriteU64P(param, varUInt64(this->limit));
pckWriteBoolP(param, this->interface.version);
pckWriteStrP(param, this->interface.versionId);
pckWritePackP(param, ioFilterGroupParamAll(ioReadFilterGroup(storageReadIo(this->read))));
pckWriteStrP(param, this->versionId);
pckWritePackP(param, ioFilterGroupParamAll(this->filterGroup));
// If the file exists
PackRead *const packRead = protocolClientSessionOpenP(this->session, .param = param);
@@ -223,11 +234,11 @@ storageReadRemoteOpen(THIS_VOID)
if (result)
{
// Clear filters since they will be run on the remote side
ioFilterGroupClear(ioReadFilterGroup(storageReadIo(this->read)));
ioFilterGroupClear(this->filterGroup);
// If the file is compressible add decompression filter locally
if (this->compressible)
ioFilterGroupAdd(ioReadFilterGroup(storageReadIo(this->read)), decompressFilterP(compressTypeLz4, .raw = true));
ioFilterGroupAdd(this->filterGroup, decompressFilterP(compressTypeLz4, .raw = true));
// Read the first block or eof
storageReadRemoteInternal(this, packRead);
@@ -259,30 +270,28 @@ storageReadRemoteClose(THIS_VOID)
}
/**********************************************************************************************************************************/
static const IoReadInterface storageReadRemoteInterface =
static const StorageReadInterface storageReadRemoteInterface =
{
.close = storageReadRemoteClose,
.eof = storageReadRemoteEof,
.filterGroup = storageReadRemoteFilterGroup,
.open = storageReadRemoteOpen,
.read = storageReadRemote,
};
FN_EXTERN StorageRead *
FN_EXTERN StorageReadRemote *
storageReadRemoteNew(
StorageRemote *const storage, ProtocolClient *const client, const String *const name, const bool ignoreMissing,
const bool compressible, const unsigned int compressLevel, const uint64_t offset, const Variant *const limit,
const bool version, const String *const versionId)
StorageRemote *const storage, ProtocolClient *const client, const String *const name, const bool compressible,
const unsigned int compressLevel, const uint64_t offset, const Variant *const limit, const String *const versionId)
{
FUNCTION_LOG_BEGIN(logLevelTrace);
FUNCTION_LOG_PARAM(STORAGE_REMOTE, storage);
FUNCTION_LOG_PARAM(PROTOCOL_CLIENT, client);
FUNCTION_LOG_PARAM(STRING, name);
FUNCTION_LOG_PARAM(BOOL, ignoreMissing);
FUNCTION_LOG_PARAM(BOOL, compressible);
FUNCTION_LOG_PARAM(UINT, compressLevel);
FUNCTION_LOG_PARAM(UINT64, offset);
FUNCTION_LOG_PARAM(VARIANT, limit);
FUNCTION_LOG_PARAM(BOOL, version);
FUNCTION_LOG_PARAM(STRING, versionId);
FUNCTION_LOG_END();
@@ -294,19 +303,19 @@ storageReadRemoteNew(
{
*this = (StorageReadRemote)
{
.interface = &storageReadRemoteInterface,
.storage = storage,
.client = client,
.session = protocolClientSessionNewP(client, PROTOCOL_COMMAND_STORAGE_READ, .async = true),
.name = strDup(name),
.compressible = compressible,
.compressLevel = compressLevel,
.offset = offset,
.limit = varDup(limit),
.versionId = strDup(versionId),
};
}
OBJ_NEW_END();
this->read = storageReadNewP(
OBJ_NAME(this, StorageRead::StorageReadRemote), STORAGE_REMOTE_TYPE, name, ignoreMissing, offset, limit,
&storageReadRemoteInterface, .version = version, .versionId = versionId);
ASSERT(this != NULL);
FUNCTION_LOG_RETURN(STORAGE_READ, this->read);
FUNCTION_LOG_RETURN(STORAGE_READ_REMOTE, this);
}

19
src/storage/remote/read.h
View File

@@ -8,11 +8,24 @@ Remote Storage Read
#include "storage/read.h"
#include "storage/remote/storage.intern.h"
/***********************************************************************************************************************************
Object type
***********************************************************************************************************************************/
typedef struct StorageReadRemote StorageReadRemote;
/***********************************************************************************************************************************
Constructors
***********************************************************************************************************************************/
FN_EXTERN StorageRead *storageReadRemoteNew(
StorageRemote *storage, ProtocolClient *client, const String *name, bool ignoreMissing, bool compressible,
unsigned int compressLevel, uint64_t offset, const Variant *limit, bool version, const String *versionId);
FN_EXTERN StorageReadRemote *storageReadRemoteNew(
StorageRemote *storage, ProtocolClient *client, const String *name, bool compressible, unsigned int compressLevel,
uint64_t offset, const Variant *limit, const String *versionId);
/***********************************************************************************************************************************
Macros for function logging
***********************************************************************************************************************************/
#define FUNCTION_LOG_STORAGE_READ_REMOTE_TYPE \
StorageReadRemote *
#define FUNCTION_LOG_STORAGE_READ_REMOTE_FORMAT(value, buffer, bufferSize) \
objNameToLog(value, "StorageReadRemote", buffer, bufferSize)
#endif

16
src/storage/remote/storage.c
View File

@@ -261,19 +261,17 @@ storageRemoteList(THIS_VOID, const String *const path, const StorageInfoLevel le
}
/**********************************************************************************************************************************/
static StorageRead *
storageRemoteNewRead(THIS_VOID, const String *const file, const bool ignoreMissing, const StorageInterfaceNewReadParam param)
static void *
storageRemoteNewRead(THIS_VOID, const String *const file, const StorageInterfaceNewReadParam param)
{
THIS(StorageRemote);
FUNCTION_LOG_BEGIN(logLevelDebug);
FUNCTION_LOG_PARAM(STORAGE_REMOTE, this);
FUNCTION_LOG_PARAM(STRING, file);
FUNCTION_LOG_PARAM(BOOL, ignoreMissing);
FUNCTION_LOG_PARAM(BOOL, param.compressible);
FUNCTION_LOG_PARAM(UINT64, param.offset);
FUNCTION_LOG_PARAM(VARIANT, param.limit);
FUNCTION_LOG_PARAM(BOOL, param.version);
FUNCTION_LOG_PARAM(STRING, param.versionId);
FUNCTION_LOG_END();
@@ -281,14 +279,14 @@ storageRemoteNewRead(THIS_VOID, const String *const file, const bool ignoreMissi
ASSERT(file != NULL);
FUNCTION_LOG_RETURN(
STORAGE_READ,
STORAGE_READ_REMOTE,
storageReadRemoteNew(
this, this->client, file, ignoreMissing, this->compressLevel > 0 ? param.compressible : false, this->compressLevel,
param.offset, param.limit, param.version, param.versionId));
this, this->client, file, this->compressLevel > 0 ? param.compressible : false, this->compressLevel,
param.offset, param.limit, param.versionId));
}
/**********************************************************************************************************************************/
static StorageWrite *
static void *
storageRemoteNewWrite(THIS_VOID, const String *const file, const StorageInterfaceNewWriteParam param)
{
THIS(StorageRemote);
@@ -313,7 +311,7 @@ storageRemoteNewWrite(THIS_VOID, const String *const file, const StorageInterfac
ASSERT(param.truncate);
FUNCTION_LOG_RETURN(
STORAGE_WRITE,
STORAGE_WRITE_REMOTE,
storageWriteRemoteNew(
this, this->client, file, param.modeFile, param.modePath, param.user, param.group, param.timeModified, param.createPath,
param.syncFile, param.syncPath, param.atomic, this->compressLevel > 0 ? param.compressible : false,

93
src/storage/remote/write.c
View File

@@ -16,28 +16,49 @@ Remote Storage File write
/***********************************************************************************************************************************
Object type
***********************************************************************************************************************************/
typedef struct StorageWriteRemote
struct StorageWriteRemote
{
StorageWriteInterface interface; // Interface
const StorageWriteInterface *interface; // Interface
StorageRemote *storage; // Storage that created this object
StorageWrite *write; // Storage write interface
ProtocolClient *client; // Protocol client to make requests with
ProtocolClientSession *session; // Protocol session for requests
IoFilterGroup *filterGroup; // Filter group
const String *name; // File name
mode_t modeFile; // File mode
mode_t modePath; // Path mode
const String *user; // User name
const String *group; // Group name
time_t timeModified; // Time modified
bool createPath; // Create path
bool syncFile; // Sync file
bool syncPath; // Sync path
bool atomic; // Atomic write
bool compressible; // Is this file compressible?
unsigned int compressLevel; // Level to use for compression
#ifdef DEBUG
uint64_t protocolWriteBytes; // How many bytes were written to the protocol layer?
#endif
} StorageWriteRemote;
};
/***********************************************************************************************************************************
Macros for function logging
Set filter group
***********************************************************************************************************************************/
#define FUNCTION_LOG_STORAGE_WRITE_REMOTE_TYPE \
StorageWriteRemote *
#define FUNCTION_LOG_STORAGE_WRITE_REMOTE_FORMAT(value, buffer, bufferSize) \
objNameToLog(value, "StorageWriteRemote", buffer, bufferSize)
static void
storageWriteRemoteFilterGroup(THIS_VOID, IoFilterGroup *const filterGroup)
{
THIS(StorageWriteRemote);
FUNCTION_TEST_BEGIN();
FUNCTION_TEST_PARAM(STORAGE_WRITE_REMOTE, this);
FUNCTION_TEST_PARAM(IO_FILTER_GROUP, filterGroup);
FUNCTION_TEST_END();
this->filterGroup = filterGroup;
FUNCTION_TEST_RETURN_VOID();
}
/***********************************************************************************************************************************
Open the file
@@ -59,33 +80,33 @@ storageWriteRemoteOpen(THIS_VOID)
if (this->compressible)
{
ioFilterGroupInsert(
ioWriteFilterGroup(storageWriteIo(this->write)), 0, decompressFilterP(compressTypeLz4, .raw = true));
this->filterGroup, 0, decompressFilterP(compressTypeLz4, .raw = true));
}
PackWrite *const param = protocolPackNew();
pckWriteStrP(param, this->interface.name);
pckWriteModeP(param, this->interface.modeFile);
pckWriteModeP(param, this->interface.modePath);
pckWriteStrP(param, this->interface.user);
pckWriteStrP(param, this->interface.group);
pckWriteTimeP(param, this->interface.timeModified);
pckWriteBoolP(param, this->interface.createPath);
pckWriteBoolP(param, this->interface.syncFile);
pckWriteBoolP(param, this->interface.syncPath);
pckWriteBoolP(param, this->interface.atomic);
pckWritePackP(param, ioFilterGroupParamAll(ioWriteFilterGroup(storageWriteIo(this->write))));
pckWriteStrP(param, this->name);
pckWriteModeP(param, this->modeFile);
pckWriteModeP(param, this->modePath);
pckWriteStrP(param, this->user);
pckWriteStrP(param, this->group);
pckWriteTimeP(param, this->timeModified);
pckWriteBoolP(param, this->createPath);
pckWriteBoolP(param, this->syncFile);
pckWriteBoolP(param, this->syncPath);
pckWriteBoolP(param, this->atomic);
pckWritePackP(param, ioFilterGroupParamAll(this->filterGroup));
protocolClientSessionOpenP(this->session, .param = param);
// Clear filters since they will be run on the remote side
ioFilterGroupClear(ioWriteFilterGroup(storageWriteIo(this->write)));
ioFilterGroupClear(this->filterGroup);
// If the file is compressible add compression filter locally
if (this->compressible)
{
ioFilterGroupAdd(
ioWriteFilterGroup(storageWriteIo(this->write)),
this->filterGroup,
compressFilterP(compressTypeLz4, (int)this->compressLevel, .raw = true));
}
}
@@ -152,7 +173,7 @@ storageWriteRemoteClose(THIS_VOID)
protocolClientSessionResponse(this->session);
ioFilterGroupResultAllSet(
ioWriteFilterGroup(storageWriteIo(this->write)), pckReadPackP(protocolClientSessionClose(this->session)));
this->filterGroup, pckReadPackP(protocolClientSessionClose(this->session)));
}
MEM_CONTEXT_TEMP_END();
@@ -163,14 +184,15 @@ storageWriteRemoteClose(THIS_VOID)
}
/**********************************************************************************************************************************/
static const IoWriteInterface storageWriteRemoteInterface =
static const StorageWriteInterface storageWriteRemoteInterface =
{
.close = storageWriteRemoteClose,
.filterGroup = storageWriteRemoteFilterGroup,
.open = storageWriteRemoteOpen,
.write = storageWriteRemote,
};
FN_EXTERN StorageWrite *
FN_EXTERN StorageWriteRemote *
storageWriteRemoteNew(
StorageRemote *const storage, ProtocolClient *const client, const String *const name, const mode_t modeFile,
const mode_t modePath, const String *const user, const String *const group, const time_t timeModified, const bool createPath,
@@ -202,20 +224,25 @@ storageWriteRemoteNew(
{
*this = (StorageWriteRemote)
{
.interface = &storageWriteRemoteInterface,
.storage = storage,
.client = client,
.session = protocolClientSessionNewP(client, PROTOCOL_COMMAND_STORAGE_WRITE, .async = true),
.name = strDup(name),
.modeFile = modeFile,
.modePath = modePath,
.user = strDup(user),
.group = strDup(group),
.timeModified = timeModified,
.createPath = createPath,
.syncFile = syncFile,
.syncPath = syncPath,
.atomic = atomic,
.compressible = compressible,
.compressLevel = compressLevel,
};
this->write = storageWriteNewP(
OBJ_NAME(this, StorageWrite::StorageWriteRemote), STORAGE_REMOTE_TYPE, name, createPath, atomic, true, syncPath,
syncFile, &storageWriteRemoteInterface, .user = user, .group = group, .modePath = modePath, .modeFile = modeFile,
.timeModified = timeModified);
}
OBJ_NEW_END();
ASSERT(this != NULL);
FUNCTION_LOG_RETURN(STORAGE_WRITE, this->write);
FUNCTION_LOG_RETURN(STORAGE_WRITE_REMOTE, this);
}

15
src/storage/remote/write.h
View File

@@ -8,12 +8,25 @@ Remote Storage File write
#include "storage/remote/storage.intern.h"
#include "storage/write.h"
/***********************************************************************************************************************************
Object type
***********************************************************************************************************************************/
typedef struct StorageWriteRemote StorageWriteRemote;
/***********************************************************************************************************************************
Constructors
***********************************************************************************************************************************/
FN_EXTERN StorageWrite *storageWriteRemoteNew(
FN_EXTERN StorageWriteRemote *storageWriteRemoteNew(
StorageRemote *storage, ProtocolClient *client, const String *name, mode_t modeFile, mode_t modePath, const String *user,
const String *group, time_t timeModified, bool createPath, bool syncFile, bool syncPath, bool atomic, bool compressible,
unsigned int compressLevel);
/***********************************************************************************************************************************
Macros for function logging
***********************************************************************************************************************************/
#define FUNCTION_LOG_STORAGE_WRITE_REMOTE_TYPE \
StorageWriteRemote *
#define FUNCTION_LOG_STORAGE_WRITE_REMOTE_FORMAT(value, buffer, bufferSize) \
objNameToLog(value, "StorageWriteRemote", buffer, bufferSize)
#endif

69
src/storage/s3/read.c
View File

@@ -18,32 +18,30 @@ Object type
#define STORAGE_READ_S3_TYPE StorageReadS3
#define STORAGE_READ_S3_PREFIX storageReadS3
typedef struct StorageReadS3
struct StorageReadS3
{
StorageReadInterface interface; // Interface
const StorageReadInterface *interface; // Interface
StorageS3 *storage; // Storage that created this object
HttpResponse *httpResponse; // HTTP response
} StorageReadS3;
const String *name; // File name
uint64_t offset; // Read offset
const Variant *limit; // Read limit (NULL for no limit)
const String *versionId; // Version id (NULL for most recent)
/***********************************************************************************************************************************
Macros for function logging
***********************************************************************************************************************************/
#define FUNCTION_LOG_STORAGE_READ_S3_TYPE \
StorageReadS3 *
#define FUNCTION_LOG_STORAGE_READ_S3_FORMAT(value, buffer, bufferSize) \
objNameToLog(value, "StorageReadS3", buffer, bufferSize)
HttpResponse *httpResponse; // HTTP response
};
/***********************************************************************************************************************************
Open the file
***********************************************************************************************************************************/
static bool
storageReadS3Open(THIS_VOID)
storageReadS3Open(THIS_VOID, const bool ignoreMissing)
{
THIS(StorageReadS3);
FUNCTION_LOG_BEGIN(logLevelTrace);
FUNCTION_LOG_PARAM(STORAGE_READ_S3, this);
FUNCTION_LOG_PARAM(BOOL, ignoreMissing);
FUNCTION_LOG_END();
ASSERT(this != NULL);
@@ -51,30 +49,24 @@ storageReadS3Open(THIS_VOID)
bool result = false;
// Read if not versioned or if versionId is not null
if (!this->interface.version || this->interface.versionId != NULL)
MEM_CONTEXT_OBJ_BEGIN(this)
{
// Request the file
MEM_CONTEXT_OBJ_BEGIN(this)
{
this->httpResponse = storageS3RequestP(
this->storage, HTTP_VERB_GET_STR, this->interface.name,
.header = httpHeaderPutRange(httpHeaderNew(NULL), this->interface.offset, this->interface.limit),
.query =
this->interface.versionId == NULL
? NULL : httpQueryPut(httpQueryNewP(), STRDEF("versionId"), this->interface.versionId),
.allowMissing = true, .contentIo = true, .sseC = true);
}
MEM_CONTEXT_OBJ_END();
this->httpResponse = storageS3RequestP(
this->storage, HTTP_VERB_GET_STR, this->name,
.header = httpHeaderPutRange(httpHeaderNew(NULL), this->offset, this->limit),
.query = this->versionId != NULL ? httpQueryPut(httpQueryNewP(), STRDEF("versionId"), this->versionId) : NULL,
.allowMissing = true, .contentIo = true, .sseC = true);
// If file exists
if (httpResponseCodeOk(this->httpResponse))
{
result = true;
}
// Else error unless ignore missing
else if (!this->interface.ignoreMissing)
THROW_FMT(FileMissingError, STORAGE_ERROR_READ_MISSING, strZ(this->interface.name));
else if (!ignoreMissing)
THROW_FMT(FileMissingError, STORAGE_ERROR_READ_MISSING, strZ(this->name));
}
MEM_CONTEXT_OBJ_END();
FUNCTION_LOG_RETURN(BOOL, result);
}
@@ -140,7 +132,7 @@ storageReadS3Close(THIS_VOID)
}
/**********************************************************************************************************************************/
static const IoReadInterface storageReadS3Interface =
static const StorageReadInterface storageReadS3Interface =
{
.close = storageReadS3Close,
.eof = storageReadS3Eof,
@@ -148,18 +140,16 @@ static const IoReadInterface storageReadS3Interface =
.read = storageReadS3,
};
FN_EXTERN StorageRead *
FN_EXTERN StorageReadS3 *
storageReadS3New(
StorageS3 *const storage, const String *const name, const bool ignoreMissing, const uint64_t offset, const Variant *const limit,
const bool version, const String *const versionId)
StorageS3 *const storage, const String *const name, const uint64_t offset, const Variant *const limit,
const String *const versionId)
{
FUNCTION_LOG_BEGIN(logLevelTrace);
FUNCTION_LOG_PARAM(STORAGE_S3, storage);
FUNCTION_LOG_PARAM(STRING, name);
FUNCTION_LOG_PARAM(BOOL, ignoreMissing);
FUNCTION_LOG_PARAM(UINT64, offset);
FUNCTION_LOG_PARAM(VARIANT, limit);
FUNCTION_LOG_PARAM(BOOL, version);
FUNCTION_LOG_PARAM(STRING, versionId);
FUNCTION_LOG_END();
@@ -171,14 +161,15 @@ storageReadS3New(
{
*this = (StorageReadS3)
{
.interface = &storageReadS3Interface,
.storage = storage,
.name = strDup(name),
.offset = offset,
.limit = varDup(limit),
.versionId = strDup(versionId),
};
}
OBJ_NEW_END();
FUNCTION_LOG_RETURN(
STORAGE_READ,
storageReadNewP(
this, STORAGE_S3_TYPE, name, ignoreMissing, offset, limit, &storageReadS3Interface, .version = version,
.versionId = versionId, .retry = true));
FUNCTION_LOG_RETURN(STORAGE_READ_S3, this);
}

18
src/storage/s3/read.h
View File

@@ -7,11 +7,23 @@ S3 Storage Read
#include "storage/read.h"
#include "storage/s3/storage.intern.h"
/***********************************************************************************************************************************
Object type
***********************************************************************************************************************************/
typedef struct StorageReadS3 StorageReadS3;
/***********************************************************************************************************************************
Constructors
***********************************************************************************************************************************/
FN_EXTERN StorageRead *storageReadS3New(
StorageS3 *storage, const String *name, bool ignoreMissing, uint64_t offset, const Variant *limit, bool version,
const String *versionId);
FN_EXTERN StorageReadS3 *storageReadS3New(
StorageS3 *storage, const String *name, uint64_t offset, const Variant *limit, const String *versionId);
/***********************************************************************************************************************************
Macros for function logging
***********************************************************************************************************************************/
#define FUNCTION_LOG_STORAGE_READ_S3_TYPE \
StorageReadS3 *
#define FUNCTION_LOG_STORAGE_READ_S3_FORMAT(value, buffer, bufferSize) \
objNameToLog(value, "StorageReadS3", buffer, bufferSize)
#endif

15
src/storage/s3/storage.c
View File

@@ -993,30 +993,27 @@ storageS3List(THIS_VOID, const String *const path, const StorageInfoLevel level,
}
/**********************************************************************************************************************************/
static StorageRead *
storageS3NewRead(THIS_VOID, const String *const file, const bool ignoreMissing, const StorageInterfaceNewReadParam param)
static void *
storageS3NewRead(THIS_VOID, const String *const file, const StorageInterfaceNewReadParam param)
{
THIS(StorageS3);
FUNCTION_LOG_BEGIN(logLevelDebug);
FUNCTION_LOG_PARAM(STORAGE_S3, this);
FUNCTION_LOG_PARAM(STRING, file);
FUNCTION_LOG_PARAM(BOOL, ignoreMissing);
FUNCTION_LOG_PARAM(UINT64, param.offset);
FUNCTION_LOG_PARAM(VARIANT, param.limit);
FUNCTION_LOG_PARAM(BOOL, param.version);
FUNCTION_LOG_PARAM(STRING, param.versionId);
FUNCTION_LOG_END();
ASSERT(this != NULL);
ASSERT(file != NULL);
FUNCTION_LOG_RETURN(
STORAGE_READ, storageReadS3New(this, file, ignoreMissing, param.offset, param.limit, param.version, param.versionId));
FUNCTION_LOG_RETURN(STORAGE_READ_S3, storageReadS3New(this, file, param.offset, param.limit, param.versionId));
}
/**********************************************************************************************************************************/
static StorageWrite *
static void *
storageS3NewWrite(THIS_VOID, const String *const file, const StorageInterfaceNewWriteParam param)
{
THIS(StorageS3);
@@ -1035,7 +1032,7 @@ storageS3NewWrite(THIS_VOID, const String *const file, const StorageInterfaceNew
ASSERT(param.group == NULL);
ASSERT(param.timeModified == 0);
FUNCTION_LOG_RETURN(STORAGE_WRITE, storageWriteS3New(this, file, this->partSize));
FUNCTION_LOG_RETURN(STORAGE_WRITE_S3, storageWriteS3New(this, file, this->partSize));
}
/**********************************************************************************************************************************/
@@ -1224,7 +1221,7 @@ storageS3Remove(THIS_VOID, const String *const file, const StorageInterfaceRemov
/**********************************************************************************************************************************/
static const StorageInterface storageInterfaceS3 =
{
.feature = 1 << storageFeatureVersioning,
.feature = 1 << storageFeatureVersioning | 1 << storageFeatureReadRetry,
.info = storageS3Info,
.list = storageS3List,

32
src/storage/s3/write.c
View File

@@ -35,25 +35,18 @@ STRING_STATIC(S3_XML_TAG_PART_NUMBER_STR, "PartNumber"
/***********************************************************************************************************************************
Object type
***********************************************************************************************************************************/
typedef struct StorageWriteS3
struct StorageWriteS3
{
StorageWriteInterface interface; // Interface
const StorageWriteInterface *interface; // Interface
StorageS3 *storage; // Storage that created this object
const String *name; // File name
HttpRequest *request; // Async request
size_t partSize;
Buffer *partBuffer;
const String *uploadId;
StringList *uploadPartList;
} StorageWriteS3;
/***********************************************************************************************************************************
Macros for function logging
***********************************************************************************************************************************/
#define FUNCTION_LOG_STORAGE_WRITE_S3_TYPE \
StorageWriteS3 *
#define FUNCTION_LOG_STORAGE_WRITE_S3_FORMAT(value, buffer, bufferSize) \
objNameToLog(value, "StorageWriteS3", buffer, bufferSize)
};
/***********************************************************************************************************************************
Flush bytes to upload part
@@ -107,7 +100,7 @@ storageWriteS3PartAsync(StorageWriteS3 *const this)
xmlDocumentNewBuf(
httpResponseContent(
storageS3RequestP(
this->storage, HTTP_VERB_POST_STR, this->interface.name,
this->storage, HTTP_VERB_POST_STR, this->name,
.query = httpQueryAdd(httpQueryNewP(), S3_QUERY_UPLOADS_STR, EMPTY_STR), .sseKms = true,
.sseC = true, .tag = true))));
@@ -128,7 +121,7 @@ storageWriteS3PartAsync(StorageWriteS3 *const this)
MEM_CONTEXT_OBJ_BEGIN(this)
{
this->request = storageS3RequestAsyncP(
this->storage, HTTP_VERB_PUT_STR, this->interface.name, .query = query, .content = this->partBuffer, .sseC = true);
this->storage, HTTP_VERB_PUT_STR, this->name, .query = query, .content = this->partBuffer, .sseC = true);
}
MEM_CONTEXT_OBJ_END();
}
@@ -225,7 +218,7 @@ storageWriteS3Close(THIS_VOID)
// Finalize the multi-part upload
HttpRequest *const request = storageS3RequestAsyncP(
this->storage, HTTP_VERB_POST_STR, this->interface.name,
this->storage, HTTP_VERB_POST_STR, this->name,
.query = httpQueryAdd(httpQueryNewP(), S3_QUERY_UPLOAD_ID_STR, this->uploadId),
.content = xmlDocumentBuf(partList));
HttpResponse *const response = storageS3ResponseP(request);
@@ -241,7 +234,7 @@ storageWriteS3Close(THIS_VOID)
else
{
storageS3RequestP(
this->storage, HTTP_VERB_PUT_STR, this->interface.name, .content = this->partBuffer, .sseKms = true,
this->storage, HTTP_VERB_PUT_STR, this->name, .content = this->partBuffer, .sseKms = true,
.sseC = true, .tag = true);
}
@@ -255,13 +248,13 @@ storageWriteS3Close(THIS_VOID)
}
/**********************************************************************************************************************************/
static const IoWriteInterface storageWriteS3Interface =
static const StorageWriteInterface storageWriteS3Interface =
{
.close = storageWriteS3Close,
.write = storageWriteS3,
};
FN_EXTERN StorageWrite *
FN_EXTERN StorageWriteS3 *
storageWriteS3New(StorageS3 *const storage, const String *const name, const size_t partSize)
{
FUNCTION_LOG_BEGIN(logLevelTrace);
@@ -277,13 +270,14 @@ storageWriteS3New(StorageS3 *const storage, const String *const name, const size
{
*this = (StorageWriteS3)
{
.interface = &storageWriteS3Interface,
.storage = storage,
.name = strDup(name),
.partSize = partSize,
.partBuffer = bufNew(0),
};
}
OBJ_NEW_END();
FUNCTION_LOG_RETURN(
STORAGE_WRITE, storageWriteNewP(this, STORAGE_S3_TYPE, name, true, true, true, true, true, &storageWriteS3Interface));
FUNCTION_LOG_RETURN(STORAGE_WRITE_S3, this);
}

15
src/storage/s3/write.h
View File

@@ -7,9 +7,22 @@ S3 Storage File Write
#include "storage/s3/storage.intern.h"
#include "storage/write.h"
/***********************************************************************************************************************************
Object type
***********************************************************************************************************************************/
typedef struct StorageWriteS3 StorageWriteS3;
/***********************************************************************************************************************************
Constructors
***********************************************************************************************************************************/
FN_EXTERN StorageWrite *storageWriteS3New(StorageS3 *storage, const String *name, size_t partSize);
FN_EXTERN StorageWriteS3 *storageWriteS3New(StorageS3 *storage, const String *name, size_t partSize);
/***********************************************************************************************************************************
Macros for function logging
***********************************************************************************************************************************/
#define FUNCTION_LOG_STORAGE_WRITE_S3_TYPE \
StorageWriteS3 *
#define FUNCTION_LOG_STORAGE_WRITE_S3_FORMAT(value, buffer, bufferSize) \
objNameToLog(value, "StorageWriteS3", buffer, bufferSize)
#endif

64
src/storage/sftp/read.c
View File

@@ -14,37 +14,34 @@ SFTP Storage Read
/***********************************************************************************************************************************
Object types
***********************************************************************************************************************************/
typedef struct StorageReadSftp
struct StorageReadSftp
{
StorageReadInterface interface; // Interface
const StorageReadInterface *interface; // Interface
StorageSftp *storage; // Storage that created this object
const String *name; // File name
uint64_t offset; // Read offset
const Variant *limit; // Read limit (NULL for no limit)
LIBSSH2_SESSION *session; // LibSsh2 session
LIBSSH2_SFTP *sftpSession; // LibSsh2 session sftp session
LIBSSH2_SFTP_HANDLE *sftpHandle; // LibSsh2 session sftp handle
LIBSSH2_SFTP_ATTRIBUTES *attr; // LibSsh2 file attributes
uint64_t current; // Current bytes read from file
bool eof; // Did we reach end of file
} StorageReadSftp;
/***********************************************************************************************************************************
Macros for function logging
***********************************************************************************************************************************/
#define FUNCTION_LOG_STORAGE_READ_SFTP_TYPE \
StorageReadSftp *
#define FUNCTION_LOG_STORAGE_READ_SFTP_FORMAT(value, buffer, bufferSize) \
objNameToLog(value, "StorageReadSftp", buffer, bufferSize)
};
/***********************************************************************************************************************************
Open the file
***********************************************************************************************************************************/
static bool
storageReadSftpOpen(THIS_VOID)
storageReadSftpOpen(THIS_VOID, const bool ignoreMissing)
{
THIS(StorageReadSftp);
FUNCTION_LOG_BEGIN(logLevelTrace);
FUNCTION_LOG_PARAM(STORAGE_READ_SFTP, this);
FUNCTION_LOG_PARAM(BOOL, ignoreMissing);
FUNCTION_LOG_END();
ASSERT(this != NULL);
@@ -52,8 +49,7 @@ storageReadSftpOpen(THIS_VOID)
do
{
this->sftpHandle = libssh2_sftp_open_ex(
this->sftpSession, strZ(this->interface.name), (unsigned int)strSize(this->interface.name), LIBSSH2_FXF_READ, 0,
LIBSSH2_SFTP_OPENFILE);
this->sftpSession, strZ(this->name), (unsigned int)strSize(this->name), LIBSSH2_FXF_READ, 0, LIBSSH2_SFTP_OPENFILE);
}
while (this->sftpHandle == NULL && storageSftpWaitFd(this->storage, libssh2_session_last_errno(this->session)));
@@ -65,14 +61,14 @@ storageReadSftpOpen(THIS_VOID)
{
if (libssh2_sftp_last_error(this->sftpSession) == LIBSSH2_FX_NO_SUCH_FILE)
{
if (!this->interface.ignoreMissing)
THROW_FMT(FileMissingError, STORAGE_ERROR_READ_MISSING, strZ(this->interface.name));
if (!ignoreMissing)
THROW_FMT(FileMissingError, STORAGE_ERROR_READ_MISSING, strZ(this->name));
}
else
{
storageSftpEvalLibSsh2Error(
rc, libssh2_sftp_last_error(this->sftpSession), &FileOpenError,
strNewFmt(STORAGE_ERROR_READ_OPEN, strZ(this->interface.name)), NULL);
strNewFmt(STORAGE_ERROR_READ_OPEN, strZ(this->name)), NULL);
}
}
}
@@ -80,8 +76,8 @@ storageReadSftpOpen(THIS_VOID)
else
{
// Seek to offset, libssh2_sftp_seek64 returns void
if (this->interface.offset != 0)
libssh2_sftp_seek64(this->sftpHandle, this->interface.offset);
if (this->offset != 0)
libssh2_sftp_seek64(this->sftpHandle, this->offset);
}
FUNCTION_LOG_RETURN(BOOL, this->sftpHandle != NULL);
@@ -111,7 +107,7 @@ storageReadSftp(THIS_VOID, Buffer *const buffer, const bool block)
{
// Rather than enable/disable limit checking just use a big number when there is no limit. We can feel pretty confident that
// no files will be > UINT64_MAX in size.
const uint64_t limit = this->interface.limit == NULL ? UINT64_MAX : varUInt64(this->interface.limit);
const uint64_t limit = this->limit == NULL ? UINT64_MAX : varUInt64(this->limit);
// Determine expected bytes to read. If remaining size in the buffer would exceed the limit then reduce the expected read.
size_t expectedBytes = bufRemains(buffer);
@@ -151,18 +147,18 @@ storageReadSftp(THIS_VOID, Buffer *const buffer, const bool block)
uint64_t sftpErr = 0;
// libssh2 sftp lseek seems to return LIBSSH2_FX_BAD_MESSAGE on a seek too far
if ((sftpErr = libssh2_sftp_last_error(this->sftpSession)) == LIBSSH2_FX_BAD_MESSAGE && this->interface.offset > 0)
THROW_FMT(FileOpenError, STORAGE_ERROR_READ_SEEK, this->interface.offset, strZ(this->interface.name));
if ((sftpErr = libssh2_sftp_last_error(this->sftpSession)) == LIBSSH2_FX_BAD_MESSAGE && this->offset > 0)
THROW_FMT(FileOpenError, STORAGE_ERROR_READ_SEEK, this->offset, strZ(this->name));
else
THROW_FMT(FileReadError, "unable to read '%s': sftp errno [%" PRIu64 "]", strZ(this->interface.name), sftpErr);
THROW_FMT(FileReadError, "unable to read '%s': sftp errno [%" PRIu64 "]", strZ(this->name), sftpErr);
}
else if (rc == LIBSSH2_ERROR_EAGAIN)
THROW_FMT(FileReadError, "timeout reading '%s'", strZ(this->interface.name));
THROW_FMT(FileReadError, "timeout reading '%s'", strZ(this->name));
else
{
storageSftpEvalLibSsh2Error(
(int)rc, libssh2_sftp_last_error(this->sftpSession), &FileReadError,
strNewFmt("unable to read '%s'", strZ(this->interface.name)), NULL);
strNewFmt("unable to read '%s'", strZ(this->name)), NULL);
}
}
@@ -208,14 +204,14 @@ storageReadSftpClose(THIS_VOID)
if (rc != LIBSSH2_ERROR_EAGAIN)
THROW_FMT(
FileCloseError,
STORAGE_ERROR_READ_CLOSE ": libssh2 errno [%d]%s", strZ(this->interface.name), rc,
STORAGE_ERROR_READ_CLOSE ": libssh2 errno [%d]%s", strZ(this->name), rc,
rc == LIBSSH2_ERROR_SFTP_PROTOCOL ?
strZ(strNewFmt(": sftp errno [%lu]", libssh2_sftp_last_error(this->sftpSession))) : "");
else
{
storageSftpEvalLibSsh2Error(
rc, libssh2_sftp_last_error(this->sftpSession), &FileCloseError,
strNewFmt("timeout closing file '%s'", strZ(this->interface.name)), NULL);
strNewFmt("timeout closing file '%s'", strZ(this->name)), NULL);
}
}
}
@@ -243,7 +239,7 @@ storageReadSftpEof(THIS_VOID)
}
/**********************************************************************************************************************************/
static const IoReadInterface storageReadSftpInterface =
static const StorageReadInterface storageReadSftpInterface =
{
.close = storageReadSftpClose,
.eof = storageReadSftpEof,
@@ -251,14 +247,13 @@ static const IoReadInterface storageReadSftpInterface =
.read = storageReadSftp,
};
FN_EXTERN StorageRead *
FN_EXTERN StorageReadSftp *
storageReadSftpNew(
StorageSftp *const storage, const String *const name, const bool ignoreMissing, LIBSSH2_SESSION *const session,
StorageSftp *const storage, const String *const name, LIBSSH2_SESSION *const session,
LIBSSH2_SFTP *const sftpSession, LIBSSH2_SFTP_HANDLE *const sftpHandle, const uint64_t offset, const Variant *const limit)
{
FUNCTION_LOG_BEGIN(logLevelTrace);
FUNCTION_LOG_PARAM(STRING, name);
FUNCTION_LOG_PARAM(BOOL, ignoreMissing);
FUNCTION_LOG_PARAM_P(VOID, session);
FUNCTION_LOG_PARAM_P(VOID, sftpSession);
FUNCTION_LOG_PARAM_P(VOID, sftpHandle);
@@ -272,16 +267,19 @@ storageReadSftpNew(
{
*this = (StorageReadSftp)
{
.interface = &storageReadSftpInterface,
.storage = storage,
.name = strDup(name),
.session = session,
.sftpSession = sftpSession,
.sftpHandle = sftpHandle,
.offset = offset,
.limit = varDup(limit),
};
}
OBJ_NEW_END();
FUNCTION_LOG_RETURN(
STORAGE_READ, storageReadNewP(this, STORAGE_SFTP_TYPE, name, ignoreMissing, offset, limit, &storageReadSftpInterface));
FUNCTION_LOG_RETURN(STORAGE_READ_SFTP, this);
}
#endif // HAVE_LIBSSH2

19
src/storage/sftp/read.h
View File

@@ -7,11 +7,24 @@ SFTP Storage Read
#include "storage/read.h"
#include "storage/sftp/storage.intern.h"
/***********************************************************************************************************************************
Object type
***********************************************************************************************************************************/
typedef struct StorageReadSftp StorageReadSftp;
/***********************************************************************************************************************************
Constructors
***********************************************************************************************************************************/
FN_EXTERN StorageRead *storageReadSftpNew(
StorageSftp *storage, const String *name, bool ignoreMissing, LIBSSH2_SESSION *session, LIBSSH2_SFTP *sftpSession,
LIBSSH2_SFTP_HANDLE *sftpHandle, uint64_t offset, const Variant *limit);
FN_EXTERN StorageReadSftp *storageReadSftpNew(
StorageSftp *storage, const String *name, LIBSSH2_SESSION *session, LIBSSH2_SFTP *sftpSession, LIBSSH2_SFTP_HANDLE *sftpHandle,
uint64_t offset, const Variant *limit);
/***********************************************************************************************************************************
Macros for function logging
***********************************************************************************************************************************/
#define FUNCTION_LOG_STORAGE_READ_SFTP_TYPE \
StorageReadSftp *
#define FUNCTION_LOG_STORAGE_READ_SFTP_FORMAT(value, buffer, bufferSize) \
objNameToLog(value, "StorageReadSftp", buffer, bufferSize)
#endif

15
src/storage/sftp/storage.c
View File

@@ -913,32 +913,29 @@ storageSftpRemove(THIS_VOID, const String *const file, const StorageInterfaceRem
}
/**********************************************************************************************************************************/
static StorageRead *
storageSftpNewRead(THIS_VOID, const String *const file, const bool ignoreMissing, const StorageInterfaceNewReadParam param)
static void *
storageSftpNewRead(THIS_VOID, const String *const file, const StorageInterfaceNewReadParam param)
{
THIS(StorageSftp);
FUNCTION_LOG_BEGIN(logLevelTrace);
FUNCTION_LOG_PARAM(STORAGE_SFTP, this);
FUNCTION_LOG_PARAM(STRING, file);
FUNCTION_LOG_PARAM(BOOL, ignoreMissing);
FUNCTION_LOG_PARAM(UINT64, param.offset);
FUNCTION_LOG_PARAM(VARIANT, param.limit);
FUNCTION_LOG_END();
ASSERT(this != NULL);
ASSERT(file != NULL);
ASSERT(!param.version);
ASSERT(param.versionId == NULL);
FUNCTION_LOG_RETURN(
STORAGE_READ,
storageReadSftpNew(
this, file, ignoreMissing, this->session, this->sftpSession, this->sftpHandle, param.offset, param.limit));
STORAGE_READ_SFTP,
storageReadSftpNew(this, file, this->session, this->sftpSession, this->sftpHandle, param.offset, param.limit));
}
/**********************************************************************************************************************************/
static StorageWrite *
static void *
storageSftpNewWrite(THIS_VOID, const String *const file, const StorageInterfaceNewWriteParam param)
{
THIS(StorageSftp);
@@ -967,7 +964,7 @@ storageSftpNewWrite(THIS_VOID, const String *const file, const StorageInterfaceN
ASSERT(param.timeModified == 0);
FUNCTION_LOG_RETURN(
STORAGE_WRITE,
STORAGE_WRITE_SFTP,
storageWriteSftpNew(
this, file, this->session, this->sftpSession, this->sftpHandle, param.modeFile, param.modePath, param.user, param.group,
param.timeModified, param.createPath, param.syncFile, param.atomic, param.truncate));

75
src/storage/sftp/write.c
View File

@@ -14,9 +14,15 @@ SFTP Storage File Write
/***********************************************************************************************************************************
Object type
***********************************************************************************************************************************/
typedef struct StorageWriteSftp
struct StorageWriteSftp
{
StorageWriteInterface interface; // Interface
const StorageWriteInterface *interface; // Interface
const String *name; // File name
mode_t modeFile; // File mode
mode_t modePath; // Path mode
bool createPath; // Create path if missing
bool syncFile; // Sync file on close
bool atomic; // Atomic write
StorageSftp *storage; // Storage that created this object
const String *nameTmp; // Temporary filename utilized for atomic ops
@@ -24,15 +30,7 @@ typedef struct StorageWriteSftp
LIBSSH2_SESSION *session; // LibSsh2 session
LIBSSH2_SFTP *sftpSession; // LibSsh2 session sftp session
LIBSSH2_SFTP_HANDLE *sftpHandle; // LibSsh2 session sftp handle
} StorageWriteSftp;
/***********************************************************************************************************************************
Macros for function logging
***********************************************************************************************************************************/
#define FUNCTION_LOG_STORAGE_WRITE_SFTP_TYPE \
StorageWriteSftp *
#define FUNCTION_LOG_STORAGE_WRITE_SFTP_FORMAT(value, buffer, bufferSize) \
objNameToLog(value, "StorageWriteSftp", buffer, bufferSize)
};
/***********************************************************************************************************************************
Open the file
@@ -55,7 +53,7 @@ storageWriteSftpOpen(THIS_VOID)
do
{
this->sftpHandle = libssh2_sftp_open_ex(
this->sftpSession, strZ(this->nameTmp), (unsigned int)strSize(this->nameTmp), flags, (int)this->interface.modeFile,
this->sftpSession, strZ(this->nameTmp), (unsigned int)strSize(this->nameTmp), flags, (int)this->modeFile,
LIBSSH2_SFTP_OPENFILE);
}
while (this->sftpHandle == NULL && storageSftpWaitFd(this->storage, libssh2_session_last_errno(this->session)));
@@ -65,13 +63,13 @@ storageWriteSftpOpen(THIS_VOID)
libssh2_sftp_last_error(this->sftpSession) == LIBSSH2_FX_NO_SUCH_FILE)
{
// Create the path
storageInterfacePathCreateP(this->storage, this->path, false, false, this->interface.modePath);
storageInterfacePathCreateP(this->storage, this->path, false, false, this->modePath);
// Open file again
do
{
this->sftpHandle = libssh2_sftp_open_ex(
this->sftpSession, strZ(this->nameTmp), (unsigned int)strSize(this->nameTmp), flags, (int)this->interface.modeFile,
this->sftpSession, strZ(this->nameTmp), (unsigned int)strSize(this->nameTmp), flags, (int)this->modeFile,
LIBSSH2_SFTP_OPENFILE);
}
while (this->sftpHandle == NULL && storageSftpWaitFd(this->storage, libssh2_session_last_errno(this->session)));
@@ -83,25 +81,25 @@ storageWriteSftpOpen(THIS_VOID)
const int rc = libssh2_session_last_errno(this->session);
if (rc == LIBSSH2_ERROR_EAGAIN)
THROW_FMT(FileOpenError, "timeout while opening file '%s'", strZ(this->interface.name));
THROW_FMT(FileOpenError, "timeout while opening file '%s'", strZ(this->name));
if (rc == LIBSSH2_ERROR_SFTP_PROTOCOL)
{
uint64_t sftpErr = libssh2_sftp_last_error(this->sftpSession);
if (sftpErr == LIBSSH2_FX_NO_SUCH_FILE)
THROW_FMT(FileMissingError, STORAGE_ERROR_WRITE_MISSING, strZ(this->interface.name));
THROW_FMT(FileMissingError, STORAGE_ERROR_WRITE_MISSING, strZ(this->name));
else
{
storageSftpEvalLibSsh2Error(
rc, sftpErr, &FileOpenError, strNewFmt(STORAGE_ERROR_WRITE_OPEN, strZ(this->interface.name)), NULL);
rc, sftpErr, &FileOpenError, strNewFmt(STORAGE_ERROR_WRITE_OPEN, strZ(this->name)), NULL);
}
}
else
{
storageSftpEvalLibSsh2Error(
rc, libssh2_sftp_last_error(this->sftpSession), &FileOpenError,
strNewFmt(STORAGE_ERROR_WRITE_OPEN, strZ(this->interface.name)), NULL);
strNewFmt(STORAGE_ERROR_WRITE_OPEN, strZ(this->name)), NULL);
}
}
@@ -181,19 +179,19 @@ storageWriteSftpUnlinkExisting(THIS_VOID)
do
{
rc = libssh2_sftp_unlink_ex(this->sftpSession, strZ(this->interface.name), (unsigned int)strSize(this->interface.name));
rc = libssh2_sftp_unlink_ex(this->sftpSession, strZ(this->name), (unsigned int)strSize(this->name));
}
while (storageSftpWaitFd(this->storage, rc));
if (rc != 0)
{
if (rc == LIBSSH2_ERROR_EAGAIN)
THROW_FMT(FileWriteError, "timeout unlinking '%s'", strZ(this->interface.name));
THROW_FMT(FileWriteError, "timeout unlinking '%s'", strZ(this->name));
else
storageSftpEvalLibSsh2Error(
rc, libssh2_sftp_last_error(this->sftpSession), &FileRemoveError,
strNewFmt("unable to remove existing '%s'", strZ(this->interface.name)), NULL);
strNewFmt("unable to remove existing '%s'", strZ(this->name)), NULL);
}
FUNCTION_LOG_RETURN_VOID();
@@ -218,8 +216,8 @@ storageWriteSftpRename(THIS_VOID)
do
{
rc = libssh2_sftp_rename_ex(
this->sftpSession, strZ(this->nameTmp), (unsigned int)strSize(this->nameTmp), strZ(this->interface.name),
(unsigned int)strSize(this->interface.name),
this->sftpSession, strZ(this->nameTmp), (unsigned int)strSize(this->nameTmp), strZ(this->name),
(unsigned int)strSize(this->name),
LIBSSH2_SFTP_RENAME_OVERWRITE | LIBSSH2_SFTP_RENAME_ATOMIC | LIBSSH2_SFTP_RENAME_NATIVE);
}
while (storageSftpWaitFd(this->storage, rc));
@@ -227,11 +225,11 @@ storageWriteSftpRename(THIS_VOID)
if (rc != 0)
{
if (rc == LIBSSH2_ERROR_EAGAIN)
THROW_FMT(FileWriteError, "timeout moving '%s' to '%s'", strZ(this->nameTmp), strZ(this->interface.name));
THROW_FMT(FileWriteError, "timeout moving '%s' to '%s'", strZ(this->nameTmp), strZ(this->name));
else
storageSftpEvalLibSsh2Error(
rc, libssh2_sftp_last_error(this->sftpSession), &FileRemoveError,
strNewFmt("unable to move '%s' to '%s'", strZ(this->nameTmp), strZ(this->interface.name)), NULL);
strNewFmt("unable to move '%s' to '%s'", strZ(this->nameTmp), strZ(this->name)), NULL);
}
FUNCTION_LOG_RETURN_VOID();
@@ -256,7 +254,7 @@ storageWriteSftpClose(THIS_VOID)
{
int rc;
if (this->interface.syncFile)
if (this->syncFile)
{
do
{
@@ -297,13 +295,13 @@ storageWriteSftpClose(THIS_VOID)
this->sftpHandle = NULL;
// Rename from temp file
if (this->interface.atomic)
if (this->atomic)
{
do
{
rc = libssh2_sftp_rename_ex(
this->sftpSession, strZ(this->nameTmp), (unsigned int)strSize(this->nameTmp), strZ(this->interface.name),
(unsigned int)strSize(this->interface.name),
this->sftpSession, strZ(this->nameTmp), (unsigned int)strSize(this->nameTmp), strZ(this->name),
(unsigned int)strSize(this->name),
LIBSSH2_SFTP_RENAME_OVERWRITE | LIBSSH2_SFTP_RENAME_ATOMIC | LIBSSH2_SFTP_RENAME_NATIVE);
}
while (storageSftpWaitFd(this->storage, rc));
@@ -327,7 +325,7 @@ storageWriteSftpClose(THIS_VOID)
{
storageSftpEvalLibSsh2Error(
rc, sftpErr, &FileCloseError,
strNewFmt("unable to move '%s' to '%s'", strZ(this->nameTmp), strZ(this->interface.name)), NULL);
strNewFmt("unable to move '%s' to '%s'", strZ(this->nameTmp), strZ(this->name)), NULL);
}
}
}
@@ -337,14 +335,14 @@ storageWriteSftpClose(THIS_VOID)
}
/**********************************************************************************************************************************/
static const IoWriteInterface storageWriteSftpInterface =
static const StorageWriteInterface storageWriteSftpInterface =
{
.close = storageWriteSftpClose,
.open = storageWriteSftpOpen,
.write = storageWriteSftp,
};
FN_EXTERN StorageWrite *
FN_EXTERN StorageWriteSftp *
storageWriteSftpNew(
StorageSftp *const storage, const String *const name, LIBSSH2_SESSION *const session, LIBSSH2_SFTP *const sftpSession,
LIBSSH2_SFTP_HANDLE *const sftpHandle, const mode_t modeFile, const mode_t modePath, const String *const user,
@@ -377,6 +375,13 @@ storageWriteSftpNew(
{
*this = (StorageWriteSftp)
{
.interface = &storageWriteSftpInterface,
.name = strDup(name),
.modeFile = modeFile,
.modePath = modePath,
.createPath = createPath,
.syncFile = syncFile,
.atomic = atomic,
.storage = storage,
.nameTmp = atomic ? strNewFmt("%s." STORAGE_FILE_TEMP_EXT, strZ(name)) : strDup(name),
.path = strPath(name),
@@ -387,11 +392,7 @@ storageWriteSftpNew(
}
OBJ_NEW_END();
FUNCTION_LOG_RETURN(
STORAGE_WRITE,
storageWriteNewP(
this, STORAGE_SFTP_TYPE, name, createPath, atomic, truncate, false, syncFile, &storageWriteSftpInterface,
.user = user, .group = group, .modePath = modePath, .modeFile = modeFile, .timeModified = timeModified));
FUNCTION_LOG_RETURN(STORAGE_WRITE_SFTP, this);
}
#endif // HAVE_LIBSSH2

15
src/storage/sftp/write.h
View File

@@ -8,12 +8,25 @@ SFTP Storage File Write
#include "storage/sftp/storage.h"
#include "storage/sftp/storage.intern.h"
/***********************************************************************************************************************************
Object type
***********************************************************************************************************************************/
typedef struct StorageWriteSftp StorageWriteSftp;
/***********************************************************************************************************************************
Constructors
***********************************************************************************************************************************/
FN_EXTERN StorageWrite *storageWriteSftpNew(
FN_EXTERN StorageWriteSftp *storageWriteSftpNew(
StorageSftp *storage, const String *name, LIBSSH2_SESSION *session, LIBSSH2_SFTP *sftpSession, LIBSSH2_SFTP_HANDLE *sftpHandle,
mode_t modeFile, mode_t modePath, const String *user, const String *group, time_t timeModified, bool createPath, bool syncFile,
bool atomic, bool truncate);
/***********************************************************************************************************************************
Macros for function logging
***********************************************************************************************************************************/
#define FUNCTION_LOG_STORAGE_WRITE_SFTP_TYPE \
StorageWriteSftp *
#define FUNCTION_LOG_STORAGE_WRITE_SFTP_FORMAT(value, buffer, bufferSize) \
objNameToLog(value, "StorageWriteSftp", buffer, bufferSize)
#endif

16
src/storage/storage.c
View File

@@ -527,9 +527,8 @@ storageNewRead(const Storage *const this, const String *const fileExp, const Sto
}
result = storageReadMove(
storageInterfaceNewReadP(
storageDriver(this), path, param.ignoreMissing, .compressible = param.compressible, .offset = param.offset,
.limit = param.limit, .version = this->targetTime != 0, .versionId = versionId),
storageReadNew(
this, path, param.ignoreMissing, param.compressible, param.offset, param.limit, this->targetTime != 0, versionId),
memContextPrior());
}
MEM_CONTEXT_TEMP_END();
@@ -567,12 +566,11 @@ storageNewWrite(const Storage *const this, const String *const fileExp, const St
MEM_CONTEXT_TEMP_BEGIN()
{
result = storageWriteMove(
storageInterfaceNewWriteP(
storageDriver(this), storagePathP(this, fileExp), .modeFile = param.modeFile != 0 ? param.modeFile : this->modeFile,
.modePath = param.modePath != 0 ? param.modePath : this->modePath, .user = param.user, .group = param.group,
.timeModified = param.timeModified, .createPath = !param.noCreatePath, .syncFile = !param.noSyncFile,
.syncPath = !param.noSyncPath, .atomic = !param.noAtomic, .truncate = !param.noTruncate,
.compressible = param.compressible),
storageWriteNew(
this, storagePathP(this, fileExp), param.modeFile != 0 ? param.modeFile : this->modeFile,
param.modePath != 0 ? param.modePath : this->modePath, param.user, param.group, param.timeModified,
!param.noCreatePath, !param.noSyncFile, !param.noSyncPath, !param.noAtomic, !param.noTruncate,
param.compressible),
memContextPrior());
}
MEM_CONTEXT_TEMP_END();

3
src/storage/storage.h
View File

@@ -60,6 +60,9 @@ typedef enum
// Does the storage support versioning?
storageFeatureVersioning,
// Can reads be retried?
storageFeatureReadRetry,
} StorageFeature;
/***********************************************************************************************************************************

17
src/storage/storage.intern.h
View File

@@ -120,19 +120,14 @@ typedef struct StorageInterfaceNewReadParam
// Limit bytes read from the file. NULL for no limit.
const Variant *limit;
// Target a specific file version. This requires a boolean as well as the versionId because file missing is indicated when the
// file is opened rather than when it is created. So if version = true and versionId = NULL then the file will be reported as
// missing on open.
bool version; // Target a file version
const String *versionId; // Id when targeting a version (NULL if version is missing)
// Target a specific file version (NULL for current version)
const String *versionId;
} StorageInterfaceNewReadParam;
typedef StorageRead *StorageInterfaceNewRead(
void *thisVoid, const String *file, bool ignoreMissing, StorageInterfaceNewReadParam param);
typedef void *StorageInterfaceNewRead(void *thisVoid, const String *file, StorageInterfaceNewReadParam param);
#define storageInterfaceNewReadP(thisVoid, file, ignoreMissing, ...) \
STORAGE_COMMON_INTERFACE(thisVoid).newRead( \
thisVoid, file, ignoreMissing, (StorageInterfaceNewReadParam){VAR_PARAM_INIT, __VA_ARGS__})
#define storageInterfaceNewReadP(thisVoid, file, ...) \
STORAGE_COMMON_INTERFACE(thisVoid).newRead(thisVoid, file, (StorageInterfaceNewReadParam){VAR_PARAM_INIT, __VA_ARGS__})
// ---------------------------------------------------------------------------------------------------------------------------------
// Create a file write object. The file should not be opened immediately -- open() will be called on the IoWrite interface when the
@@ -171,7 +166,7 @@ typedef struct StorageInterfaceNewWriteParam
bool compressible;
} StorageInterfaceNewWriteParam;
typedef StorageWrite *StorageInterfaceNewWrite(void *thisVoid, const String *file, StorageInterfaceNewWriteParam param);
typedef void *StorageInterfaceNewWrite(void *thisVoid, const String *file, StorageInterfaceNewWriteParam param);
#define storageInterfaceNewWriteP(thisVoid, file, ...) \
STORAGE_COMMON_INTERFACE(thisVoid).newWrite(thisVoid, file, (StorageInterfaceNewWriteParam){VAR_PARAM_INIT, __VA_ARGS__})

196
src/storage/write.c
View File

@@ -14,78 +14,188 @@ Object type
struct StorageWrite
{
StorageWritePub pub; // Publicly accessible variables
void *driver; // Driver
};
/***********************************************************************************************************************************
Macros for function logging
Open the file
***********************************************************************************************************************************/
#define FUNCTION_LOG_STORAGE_WRITE_INTERFACE_TYPE \
StorageWriteInterface
#define FUNCTION_LOG_STORAGE_WRITE_INTERFACE_FORMAT(value, buffer, bufferSize) \
objNameToLog(&value, "StorageWriteInterface", buffer, bufferSize)
static void
storageWriteOpen(THIS_VOID)
{
THIS(StorageWrite);
FUNCTION_LOG_BEGIN(logLevelTrace);
FUNCTION_LOG_PARAM(STORAGE_WRITE, this);
FUNCTION_LOG_END();
ASSERT(this != NULL);
storageWriteDriverInterface(this->driver)->open(this->driver);
FUNCTION_LOG_RETURN_VOID();
}
/***********************************************************************************************************************************
This object expects its context to be created in advance. This is so the calling function can add whatever data it wants without
required multiple functions and contexts to make it safe.
Write to a file
***********************************************************************************************************************************/
static void
storageWrite(THIS_VOID, const Buffer *const buffer)
{
THIS(StorageWrite);
FUNCTION_LOG_BEGIN(logLevelTrace);
FUNCTION_LOG_PARAM(STORAGE_WRITE, this);
FUNCTION_LOG_PARAM(BUFFER, buffer);
FUNCTION_LOG_END();
ASSERT(this != NULL);
storageWriteDriverInterface(this->driver)->write(this->driver, buffer);
FUNCTION_LOG_RETURN_VOID();
}
/***********************************************************************************************************************************
Close the file
***********************************************************************************************************************************/
static void
storageWriteClose(THIS_VOID)
{
THIS(StorageWrite);
FUNCTION_LOG_BEGIN(logLevelTrace);
FUNCTION_LOG_PARAM(STORAGE_WRITE, this);
FUNCTION_LOG_END();
ASSERT(this != NULL);
storageWriteDriverInterface(this->driver)->close(this->driver);
FUNCTION_LOG_RETURN_VOID();
}
/***********************************************************************************************************************************
Get file descriptor
***********************************************************************************************************************************/
static int
storageWriteFd(const THIS_VOID)
{
THIS(const StorageWrite);
FUNCTION_TEST_BEGIN();
FUNCTION_TEST_PARAM(STORAGE_WRITE, this);
FUNCTION_TEST_END();
ASSERT(this != NULL);
FUNCTION_TEST_RETURN(INT, storageWriteDriverInterface(this->driver)->fd(this->driver));
}
/***********************************************************************************************************************************
Seek in file
***********************************************************************************************************************************/
static void
storageWriteSeek(THIS_VOID, const uint64_t position)
{
THIS(StorageWrite);
FUNCTION_LOG_BEGIN(logLevelTrace);
FUNCTION_LOG_PARAM(STORAGE_WRITE, this);
FUNCTION_LOG_PARAM(UINT64, position);
FUNCTION_LOG_END();
ASSERT(this != NULL);
storageWriteDriverInterface(this->driver)->seek(this->driver, position);
FUNCTION_LOG_RETURN_VOID();
}
/**********************************************************************************************************************************/
static const IoWriteInterface storageIoWriteInterface =
{
.close = storageWriteClose,
.fd = storageWriteFd,
.open = storageWriteOpen,
.seek = storageWriteSeek,
.write = storageWrite,
};
FN_EXTERN StorageWrite *
storageWriteNew(
void *const driver, const StringId type, const String *const name, const bool createPath, const bool atomic,
const bool truncate, const bool syncPath, const bool syncFile, const IoWriteInterface *const ioInterface,
const StorageWriteNewParam param)
const Storage *const storage, const String *const name, const mode_t modeFile, const mode_t modePath, const String *const user,
const String *const group, const time_t timeModified, const bool createPath, const bool syncFile, const bool syncPath,
const bool atomic, const bool truncate, const bool compressible)
{
FUNCTION_LOG_BEGIN(logLevelTrace);
FUNCTION_LOG_PARAM_P(VOID, driver);
FUNCTION_LOG_PARAM(STRING_ID, type);
FUNCTION_LOG_PARAM(STORAGE, storage);
FUNCTION_LOG_PARAM(STRING, name);
FUNCTION_LOG_PARAM(MODE, modeFile);
FUNCTION_LOG_PARAM(MODE, modePath);
FUNCTION_LOG_PARAM(STRING, user);
FUNCTION_LOG_PARAM(STRING, group);
FUNCTION_LOG_PARAM(TIME, timeModified);
FUNCTION_LOG_PARAM(BOOL, createPath);
FUNCTION_LOG_PARAM(BOOL, syncFile);
FUNCTION_LOG_PARAM(BOOL, syncPath);
FUNCTION_LOG_PARAM(BOOL, atomic);
FUNCTION_LOG_PARAM(BOOL, truncate);
FUNCTION_LOG_PARAM(BOOL, syncPath);
FUNCTION_LOG_PARAM(BOOL, syncFile);
FUNCTION_LOG_PARAM_P(IO_WRITE_INTERFACE, ioInterface);
FUNCTION_LOG_PARAM(STRING, param.user);
FUNCTION_LOG_PARAM(STRING, param.group);
FUNCTION_LOG_PARAM(UINT, param.modePath);
FUNCTION_LOG_PARAM(UINT, param.modeFile);
FUNCTION_LOG_PARAM(TIME, param.timeModified);
FUNCTION_LOG_PARAM(BOOL, compressible);
FUNCTION_LOG_END();
FUNCTION_AUDIT_HELPER();
ASSERT(driver != NULL);
ASSERT(storage != NULL);
ASSERT(name != NULL);
ASSERT(ioInterface != NULL);
ASSERT(ioInterface->close != NULL);
ASSERT(ioInterface->write != NULL);
OBJ_NEW_BEGIN(StorageWrite, .childQty = MEM_CONTEXT_QTY_MAX)
{
// Initialize the interface
*((StorageWriteInterface *)driver) = (StorageWriteInterface)
{
.name = strDup(name),
.createPath = createPath,
.atomic = atomic,
.truncate = truncate,
.syncPath = syncPath,
.syncFile = syncFile,
.user = strDup(param.user),
.group = strDup(param.group),
.modePath = param.modePath,
.modeFile = param.modeFile,
.timeModified = param.timeModified,
};
*this = (StorageWrite)
{
.driver = storageInterfaceNewWriteP(
storageDriver(storage), name, .modeFile = modeFile, .modePath = modePath, .user = user, .group = group,
.timeModified = timeModified, .createPath = createPath, .syncFile = syncFile, .syncPath = syncPath,
.atomic = atomic, .truncate = truncate, .compressible = compressible),
.pub =
{
.interface = driver,
.io = ioWriteNew(driver, *ioInterface),
.type = type,
.type = storageType(storage),
.name = strDup(name),
.createPath = createPath,
.atomic = atomic,
.truncate = truncate,
.syncPath = syncPath,
.syncFile = syncFile,
.user = strDup(user),
.group = strDup(group),
.modePath = modePath,
.modeFile = modeFile,
.timeModified = timeModified,
},
};
ASSERT(storageWriteDriverInterface(this->driver)->close != NULL);
ASSERT(storageWriteDriverInterface(this->driver)->write != NULL);
// Remove fd method if it does not exist in the driver
IoWriteInterface storageIoWriteInterfaceCopy = storageIoWriteInterface;
if (storageWriteDriverInterface(this->driver)->fd == NULL)
storageIoWriteInterfaceCopy.fd = NULL;
// Remove open method if it does not exist in the driver (cloud drivers don't have open)
if (storageWriteDriverInterface(this->driver)->open == NULL)
storageIoWriteInterfaceCopy.open = NULL;
// Remove seek method if it does not exist in the driver
if (storageWriteDriverInterface(this->driver)->seek == NULL)
storageIoWriteInterfaceCopy.seek = NULL;
this->pub.io = ioWriteNew(this, storageIoWriteInterfaceCopy);
// Set filter group when interface function exists
if (storageWriteDriverInterface(this->driver)->filterGroup != NULL)
storageWriteDriverInterface(this->driver)->filterGroup(this->driver, ioWriteFilterGroup(storageWriteIo(this)));
}
OBJ_NEW_END();

34
src/storage/write.h
View File

@@ -10,7 +10,6 @@ Object type
typedef struct StorageWrite StorageWrite;
#include "common/io/write.h"
#include "common/type/buffer.h"
#include "common/type/object.h"
#include "common/type/string.h"
#include "common/type/stringId.h"
@@ -29,19 +28,36 @@ storageWriteMove(StorageWrite *const this, MemContext *const parentNew)
/***********************************************************************************************************************************
Getters/Setters
***********************************************************************************************************************************/
typedef struct StorageWritePub
{
StringId type; // Storage type
const String *name; // File name
bool createPath; // Will the path be created if required?
bool atomic; // Are writes atomic?
bool truncate; // Truncate file if it exists
bool syncPath; // Will path be synced?
bool syncFile; // Will file be synced?
const String *user; // Set user ownership
const String *group; // Set group ownership
mode_t modePath; // Path mode
mode_t modeFile; // File mode
time_t timeModified; // Set modification time
IoWrite *io; // Write interface
} StorageWritePub;
// Will the file be written atomically? Atomic writes means the file will be complete or be missing. Filesystems have different ways
// to accomplish this.
FN_INLINE_ALWAYS bool
storageWriteAtomic(const StorageWrite *const this)
{
return storageWriteInterface(this)->atomic;
return THIS_PUB(StorageWrite)->atomic;
}
// Will the path be created if required?
FN_INLINE_ALWAYS bool
storageWriteCreatePath(const StorageWrite *const this)
{
return storageWriteInterface(this)->createPath;
return THIS_PUB(StorageWrite)->createPath;
}
// Write interface
@@ -55,42 +71,42 @@ storageWriteIo(const StorageWrite *const this)
FN_INLINE_ALWAYS mode_t
storageWriteModeFile(const StorageWrite *const this)
{
return storageWriteInterface(this)->modeFile;
return THIS_PUB(StorageWrite)->modeFile;
}
// Path mode (if the destination path needs to be create)
FN_INLINE_ALWAYS mode_t
storageWriteModePath(const StorageWrite *const this)
{
return storageWriteInterface(this)->modePath;
return THIS_PUB(StorageWrite)->modePath;
}
// File name
FN_INLINE_ALWAYS const String *
storageWriteName(const StorageWrite *const this)
{
return storageWriteInterface(this)->name;
return THIS_PUB(StorageWrite)->name;
}
// Will the file be synced before it is closed?
FN_INLINE_ALWAYS bool
storageWriteSyncFile(const StorageWrite *const this)
{
return storageWriteInterface(this)->syncFile;
return THIS_PUB(StorageWrite)->syncFile;
}
// Will the path be synced after the file is closed?
FN_INLINE_ALWAYS bool
storageWriteSyncPath(const StorageWrite *const this)
{
return storageWriteInterface(this)->syncPath;
return THIS_PUB(StorageWrite)->syncPath;
}
// Will the file be truncated if it exists?
FN_INLINE_ALWAYS bool
storageWriteTruncate(const StorageWrite *const this)
{
return storageWriteInterface(this)->truncate;
return THIS_PUB(StorageWrite)->truncate;
}
// File type

56
src/storage/write.intern.h
View File

@@ -5,6 +5,7 @@ Storage Write Interface Internal
#define STORAGE_WRITE_INTERN_H
#include "common/io/write.h"
#include "storage/storage.h"
#include "version.h"
/***********************************************************************************************************************************
@@ -13,57 +14,32 @@ Temporary file extension
#define STORAGE_FILE_TEMP_EXT PROJECT_BIN ".tmp"
/***********************************************************************************************************************************
Constructors
Interface
***********************************************************************************************************************************/
typedef struct StorageWriteInterface
{
const String *name; // File name
bool createPath; // Will the path be created if required?
bool atomic; // Are writes atomic?
bool truncate; // Truncate file if it exists
bool syncPath; // Will path be synced?
bool syncFile; // Will file be synced?
const String *user; // Set user ownership
const String *group; // Set group ownership
mode_t modePath; // Path mode
mode_t modeFile; // File mode
time_t timeModified; // Set modification time
void (*filterGroup)(void *driver, IoFilterGroup *filterGroup); // Set filter group (optional)
void (*open)(void *driver); // Open write
void (*write)(void *driver, const Buffer *buffer); // Write bytes
void (*seek)(void *driver, uint64_t position); // Seek (optional)
int (*fd)(const void *driver); // Write file descriptor (optional)
void (*close)(void *driver); // Close write
} StorageWriteInterface;
typedef struct StorageWriteNewParam
{
VAR_PARAM_HEADER;
const String *user; // Set user ownership
const String *group; // Set group ownership
mode_t modePath; // Path mode
mode_t modeFile; // File mode
time_t timeModified; // Set modification time
} StorageWriteNewParam;
#define storageWriteNewP(driver, type, name, createPath, atomic, truncate, syncPath, syncFile, ioInterface, ...) \
storageWriteNew( \
driver, type, name, createPath, atomic, truncate, syncPath, syncFile, ioInterface, \
(StorageWriteNewParam){VAR_PARAM_INIT, __VA_ARGS__})
/***********************************************************************************************************************************
Constructors
***********************************************************************************************************************************/
FN_EXTERN StorageWrite *storageWriteNew(
void *driver, StringId type, const String *name, bool createPath, bool atomic, bool truncate, bool syncPath, bool syncFile,
const IoWriteInterface *ioInterface, StorageWriteNewParam param);
const Storage *storage, const String *name, mode_t modeFile, mode_t modePath, const String *user, const String *group,
time_t timeModified, bool createPath, bool syncFile, bool syncPath, bool atomic, bool truncate, bool compressible);
/***********************************************************************************************************************************
Getters/Setters
Getters
***********************************************************************************************************************************/
typedef struct StorageWritePub
{
const StorageWriteInterface *interface; // File data (name, driver type, etc.)
IoWrite *io; // Write interface
StringId type; // Storage type
} StorageWritePub;
// Write interface
FN_INLINE_ALWAYS const StorageWriteInterface *
storageWriteInterface(const StorageWrite *const this)
storageWriteDriverInterface(const void *const driver)
{
return THIS_PUB(StorageWrite)->interface;
return *(const StorageWriteInterface *const *)driver;
}
#endif

103
test/src/common/harnessStorageHelper.c
View File

@@ -28,15 +28,15 @@ typedef struct HrnStorageTest
typedef struct HrnStorageWriteTest
{
StorageWriteInterface interface; // Interface
IoWrite *base; // Posix IO for base file
IoWrite *version; // Posix IO for version file
const StorageWriteInterface *interface; // Interface
void *base; // Posix driver for base file
void *version; // Posix driver for version file
} HrnStorageWriteTest;
typedef struct HrnStorageReadTest
{
StorageReadInterface interface; // Interface
IoRead *posix; // Posix IO for the file
const StorageReadInterface *interface; // Interface
void *driver; // Read driver
} HrnStorageReadTest;
/***********************************************************************************************************************************
@@ -120,21 +120,18 @@ hrnStorageTestVersionFind(const Storage *const storage, const String *const file
Test storage read driver
***********************************************************************************************************************************/
static bool
hrnStorageReadTestOpen(THIS_VOID)
hrnStorageReadTestOpen(THIS_VOID, const bool ignoreMissing)
{
THIS(HrnStorageReadTest);
FUNCTION_HARNESS_BEGIN();
FUNCTION_HARNESS_PARAM(HRN_STORAGE_READ_TEST, this);
FUNCTION_HARNESS_PARAM(BOOL, ignoreMissing);
FUNCTION_HARNESS_END();
ASSERT(this != NULL);
// If the version is missing
if (this->interface.version && this->interface.versionId == NULL)
FUNCTION_HARNESS_RETURN(BOOL, false);
FUNCTION_HARNESS_RETURN(BOOL, ioReadInterface(this->posix)->open(ioReadDriver(this->posix)));
FUNCTION_HARNESS_RETURN(BOOL, storageReadDriverInterface(this->driver)->open(this->driver, ignoreMissing));
}
static size_t
@@ -151,7 +148,7 @@ hrnStorageReadTest(THIS_VOID, Buffer *const buffer, const bool block)
ASSERT(this != NULL);
ASSERT(buffer != NULL && !bufFull(buffer));
FUNCTION_HARNESS_RETURN(SIZE, ioReadInterface(this->posix)->read(ioReadDriver(this->posix), buffer, block));
FUNCTION_HARNESS_RETURN(SIZE, storageReadDriverInterface(this->driver)->read(this->driver, buffer, block));
}
static void
@@ -163,7 +160,7 @@ hrnStorageReadTestClose(THIS_VOID)
FUNCTION_HARNESS_PARAM(HRN_STORAGE_READ_TEST, this);
FUNCTION_HARNESS_END();
ioReadInterface(this->posix)->close(ioReadDriver(this->posix));
storageReadDriverInterface(this->driver)->close(this->driver);
FUNCTION_HARNESS_RETURN_VOID();
}
@@ -179,7 +176,7 @@ hrnStorageReadTestEof(THIS_VOID)
ASSERT(this != NULL);
FUNCTION_TEST_RETURN(BOOL, ioReadInterface(this->posix)->eof(ioReadDriver(this->posix)));
FUNCTION_TEST_RETURN(BOOL, storageReadDriverInterface(this->driver)->eof(this->driver));
}
static int
@@ -193,10 +190,10 @@ hrnStorageReadTestFd(const THIS_VOID)
ASSERT(this != NULL);
FUNCTION_TEST_RETURN(INT, ioReadInterface(this->posix)->fd(ioReadDriver(this->posix)));
FUNCTION_TEST_RETURN(INT, storageReadDriverInterface(this->driver)->fd(this->driver));
}
static const IoReadInterface hrnStorageReadTestInterface =
static const StorageReadInterface hrnStorageReadTestInterface =
{
.close = hrnStorageReadTestClose,
.eof = hrnStorageReadTestEof,
@@ -205,17 +202,15 @@ static const IoReadInterface hrnStorageReadTestInterface =
.read = hrnStorageReadTest,
};
static StorageRead *
static HrnStorageReadTest *
hrnStorageReadTestNew(
StoragePosix *const storage, const String *name, const bool ignoreMissing, const uint64_t offset,
const Variant *const limit, const bool version, const String *const versionId)
StoragePosix *const storage, const String *name, const uint64_t offset, const Variant *const limit,
const String *const versionId)
{
FUNCTION_HARNESS_BEGIN();
FUNCTION_HARNESS_PARAM(STRING, name);
FUNCTION_HARNESS_PARAM(BOOL, ignoreMissing);
FUNCTION_HARNESS_PARAM(UINT64, offset);
FUNCTION_HARNESS_PARAM(VARIANT, limit);
FUNCTION_HARNESS_PARAM(BOOL, version);
FUNCTION_HARNESS_PARAM(STRING, versionId);
FUNCTION_HARNESS_END();
@@ -229,16 +224,13 @@ hrnStorageReadTestNew(
*this = (HrnStorageReadTest)
{
.posix = storageReadIo(storageReadPosixNew(storage, name, ignoreMissing, offset, limit)),
.interface = &hrnStorageReadTestInterface,
.driver = storageReadPosixNew(storage, name, offset, limit),
};
}
OBJ_NEW_END();
FUNCTION_HARNESS_RETURN(
STORAGE_READ,
storageReadNewP(
this, STORAGE_TEST_TYPE, name, ignoreMissing, offset, limit, &hrnStorageReadTestInterface, .version = version,
.versionId = versionId));
FUNCTION_HARNESS_RETURN(HRN_STORAGE_READ_TEST, this);
}
/***********************************************************************************************************************************
@@ -255,8 +247,8 @@ hrnStorageWriteTestOpen(THIS_VOID)
ASSERT(this != NULL);
ioWriteInterface(this->base)->open(ioWriteDriver(this->base));
ioWriteInterface(this->version)->open(ioWriteDriver(this->version));
storageWriteDriverInterface(this->base)->open(this->base);
storageWriteDriverInterface(this->version)->open(this->version);
FUNCTION_HARNESS_RETURN_VOID();
}
@@ -274,8 +266,8 @@ hrnStorageWriteTest(THIS_VOID, const Buffer *const buffer)
ASSERT(this != NULL);
ASSERT(buffer != NULL);
ioWriteInterface(this->base)->write(ioWriteDriver(this->base), buffer);
ioWriteInterface(this->version)->write(ioWriteDriver(this->version), buffer);
storageWriteDriverInterface(this->base)->write(this->base, buffer);
storageWriteDriverInterface(this->version)->write(this->version, buffer);
FUNCTION_HARNESS_RETURN_VOID();
}
@@ -291,8 +283,8 @@ hrnStorageWriteTestClose(THIS_VOID)
ASSERT(this != NULL);
ioWriteInterface(this->base)->close(ioWriteDriver(this->base));
ioWriteInterface(this->version)->close(ioWriteDriver(this->version));
storageWriteDriverInterface(this->base)->close(this->base);
storageWriteDriverInterface(this->version)->close(this->version);
FUNCTION_HARNESS_RETURN_VOID();
}
@@ -308,10 +300,10 @@ hrnStorageWriteTestFd(const THIS_VOID)
ASSERT(this != NULL);
FUNCTION_HARNESS_RETURN(INT, ioWriteInterface(this->base)->fd(ioWriteDriver(this->base)));
FUNCTION_HARNESS_RETURN(INT, storageWriteDriverInterface(this->base)->fd(this->base));
}
static const IoWriteInterface hrnStorageWriteTestInterface =
static const StorageWriteInterface hrnStorageWriteTestInterface =
{
.close = hrnStorageWriteTestClose,
.fd = hrnStorageWriteTestFd,
@@ -319,7 +311,7 @@ static const IoWriteInterface hrnStorageWriteTestInterface =
.write = hrnStorageWriteTest,
};
static StorageWrite *
static HrnStorageWriteTest *
hrnStorageWriteTestNew(
Storage *const storagePosix, const String *const name, const mode_t modeFile, const mode_t modePath,
const String *const user, const String *const group, time_t timeModified, const bool createPath, const bool syncFile,
@@ -355,23 +347,18 @@ hrnStorageWriteTestNew(
*this = (HrnStorageWriteTest)
{
.base = storageWriteIo(
storageWritePosixNew(
storageDriver(storagePosix), name, modeFile, modePath, user, group, timeModified, createPath, false, false,
false, truncate)),
.version = storageWriteIo(
storageWritePosixNew(
storageDriver(storagePosix), hrnStorageTestVersionFind(storagePosix, name), modeFile, modePath, user, group,
timeModified, createPath, false, false, false, truncate)),
.interface = &hrnStorageWriteTestInterface,
.base = storageWritePosixNew(
storageDriver(storagePosix), name, modeFile, modePath, user, group, timeModified, createPath, false, false,
false, truncate),
.version = storageWritePosixNew(
storageDriver(storagePosix), hrnStorageTestVersionFind(storagePosix, name), modeFile, modePath, user, group,
timeModified, createPath, false, false, false, truncate),
};
}
OBJ_NEW_END();
FUNCTION_HARNESS_RETURN(
STORAGE_WRITE,
storageWriteNewP(
this, STORAGE_TEST_TYPE, name, createPath, false, truncate, false, false, &hrnStorageWriteTestInterface,
.user = user, .group = group, .modePath = modePath, .modeFile = modeFile, .timeModified = timeModified));
FUNCTION_HARNESS_RETURN(HRN_STORAGE_WRITE_TEST, this);
}
/***********************************************************************************************************************************
@@ -487,18 +474,16 @@ hrnStorageTestList(THIS_VOID, const String *const path, const StorageInfoLevel l
FUNCTION_HARNESS_RETURN(STORAGE_LIST, result);
}
static StorageRead *
hrnStorageTestNewRead(THIS_VOID, const String *file, const bool ignoreMissing, const StorageInterfaceNewReadParam param)
static void *
hrnStorageTestNewRead(THIS_VOID, const String *file, const StorageInterfaceNewReadParam param)
{
THIS(HrnStorageTest);
FUNCTION_HARNESS_BEGIN();
FUNCTION_HARNESS_PARAM(HRN_STORAGE_TEST, this);
FUNCTION_HARNESS_PARAM(STRING, file);
FUNCTION_HARNESS_PARAM(BOOL, ignoreMissing);
FUNCTION_HARNESS_PARAM(UINT64, param.offset);
FUNCTION_HARNESS_PARAM(VARIANT, param.limit);
FUNCTION_HARNESS_PARAM(BOOL, param.version);
FUNCTION_HARNESS_PARAM(STRING, param.versionId);
FUNCTION_HARNESS_END();
@@ -507,12 +492,11 @@ hrnStorageTestNewRead(THIS_VOID, const String *file, const bool ignoreMissing, c
hrnStorageTestSecretCheck(file);
FUNCTION_HARNESS_RETURN(
STORAGE_READ,
hrnStorageReadTestNew(
storageDriver(this->storagePosix), file, ignoreMissing, param.offset, param.limit, param.version, param.versionId));
HRN_STORAGE_READ_TEST,
hrnStorageReadTestNew(storageDriver(this->storagePosix), file, param.offset, param.limit, param.versionId));
}
static StorageWrite *
static void *
hrnStorageTestNewWrite(THIS_VOID, const String *const file, const StorageInterfaceNewWriteParam param)
{
THIS(HrnStorageTest);
@@ -537,7 +521,7 @@ hrnStorageTestNewWrite(THIS_VOID, const String *const file, const StorageInterfa
hrnStorageTestSecretCheck(file);
FUNCTION_HARNESS_RETURN(
STORAGE_WRITE,
HRN_STORAGE_WRITE_TEST,
hrnStorageWriteTestNew(
this->storagePosix, file, param.modeFile, param.modePath, param.user, param.group, param.timeModified,
param.createPath, param.syncFile, param.syncPath, param.atomic, param.truncate));
@@ -667,7 +651,8 @@ hrnStorageTestNew(
static const StorageInterface hrnStorageInterfaceTest =
{
.feature = 1 << storageFeaturePath | 1 << storageFeatureInfoDetail | 1 << storageFeatureVersioning,
.feature =
1 << storageFeaturePath | 1 << storageFeaturePathSync | 1 << storageFeatureInfoDetail | 1 << storageFeatureVersioning,
.info = hrnStorageTestInfo,
.list = hrnStorageTestList,

10
test/src/module/storage/azureTest.c
View File

@@ -654,7 +654,7 @@ testRun(void)
TEST_RESULT_BOOL(storageWriteSyncPath(write), true, "path is synced");
TEST_RESULT_BOOL(storageWriteTruncate(write), true, "file will be truncated");
TEST_RESULT_VOID(storageWriteAzureClose(ioWriteDriver(storageWriteIo(write))), "close file again");
TEST_RESULT_VOID(storageWriteAzureClose(write->driver), "close file again");
// -----------------------------------------------------------------------------------------------------------------
TEST_TITLE("write zero-length file");
@@ -1329,15 +1329,11 @@ testRun(void)
ioWriteOpen(storageWriteIo(write));
ioWrite(storageWriteIo(write), BUFSTRDEF("123456789012345678"));
TEST_RESULT_VOID(
bufResize(((StorageWriteAzure *)ioWriteDriver(storageWriteIo(write)))->blockBuffer, 17),
"resize part buffer to 17");
TEST_RESULT_VOID(bufResize(((StorageWriteAzure *)write->driver)->blockBuffer, 17), "resize part buffer to 17");
ioWrite(storageWriteIo(write), BUFSTRDEF("90"));
TEST_RESULT_UINT(
((StorageWriteAzure *)ioWriteDriver(storageWriteIo(write)))->blockSize, 16,
"part buffer reset to 16 (default)");
TEST_RESULT_UINT(((StorageWriteAzure *)write->driver)->blockSize, 16, "part buffer reset to 16 (default)");
ioWriteClose(storageWriteIo(write));
ioBufferSizeSet(ioBufferSizeDefault);

9
test/src/module/storage/gcsTest.c
View File

@@ -605,7 +605,7 @@ testRun(void)
TEST_RESULT_BOOL(storageWriteSyncPath(write), true, "path is synced");
TEST_RESULT_BOOL(storageWriteTruncate(write), true, "file will be truncated");
TEST_RESULT_VOID(storageWriteGcsClose(ioWriteDriver(storageWriteIo(write))), "close file again");
TEST_RESULT_VOID(storageWriteGcsClose(write->driver), "close file again");
// -----------------------------------------------------------------------------------------------------------------
TEST_TITLE("write zero-length file");
@@ -1410,14 +1410,11 @@ testRun(void)
ioWriteOpen(storageWriteIo(write));
ioWrite(storageWriteIo(write), BUFSTRDEF("123456789012345678"));
TEST_RESULT_VOID(
bufResize(((StorageWriteGcs *)ioWriteDriver(storageWriteIo(write)))->chunkBuffer, 17),
"resize part buffer to 17");
TEST_RESULT_VOID(bufResize(((StorageWriteGcs *)write->driver)->chunkBuffer, 17), "resize part buffer to 17");
ioWrite(storageWriteIo(write), BUFSTRDEF("90"));
TEST_RESULT_UINT(
((StorageWriteGcs *)ioWriteDriver(storageWriteIo(write)))->chunkSize, 16, "part buffer reset to 16 (default)");
TEST_RESULT_UINT(((StorageWriteGcs *)write->driver)->chunkSize, 16, "part buffer reset to 16 (default)");
ioWriteClose(storageWriteIo(write));
ioBufferSizeSet(ioBufferSizeDefault);

21
test/src/module/storage/posixTest.c
View File

@@ -995,7 +995,7 @@ testRun(void)
"new write file (defaults)");
TEST_RESULT_VOID(ioWriteOpen(storageWriteIo(file)), "open file");
TEST_RESULT_INT(
ioWriteFd(storageWriteIo(file)), ((StorageWritePosix *)ioWriteDriver(storageWriteIo(file)))->fd, "check write fd");
ioWriteFd(storageWriteIo(file)), ((StorageWritePosix *)file->driver)->fd, "check write fd");
TEST_RESULT_VOID(ioWriteClose(storageWriteIo(file)), "close file");
TEST_RESULT_INT(storageInfoP(storageTest, strPath(fileName)).mode, 0750, "check path mode");
TEST_RESULT_INT(storageInfoP(storageTest, fileName).mode, 0640, "check file mode");
@@ -1042,7 +1042,7 @@ testRun(void)
"new write file (set mode)");
TEST_RESULT_VOID(ioWriteOpen(storageWriteIo(file)), "open file");
TEST_RESULT_VOID(ioWriteClose(storageWriteIo(file)), "close file");
TEST_RESULT_VOID(storageWritePosixClose(ioWriteDriver(storageWriteIo(file))), "close file again");
TEST_RESULT_VOID(storageWritePosixClose(file->driver), "close file again");
TEST_RESULT_INT(storageInfoP(storageTest, strPath(fileName)).mode, 0700, "check path mode");
TEST_RESULT_INT(storageInfoP(storageTest, fileName).mode, 0600, "check file mode");
@@ -1366,25 +1366,25 @@ testRun(void)
TEST_RESULT_VOID(ioWriteOpen(storageWriteIo(file)), "open file");
// Close the file descriptor so operations will fail
close(((StorageWritePosix *)ioWriteDriver(storageWriteIo(file)))->fd);
close(((StorageWritePosix *)file->driver)->fd);
storageRemoveP(storageTest, fileTmp, .errorOnMissing = true);
TEST_ERROR_FMT(
storageWritePosix(ioWriteDriver(storageWriteIo(file)), buffer), FileWriteError,
storageWritePosix(file->driver, buffer), FileWriteError,
"unable to write '%s.pgbackrest.tmp': [9] Bad file descriptor", strZ(fileName));
TEST_ERROR_FMT(
storageWritePosixClose(ioWriteDriver(storageWriteIo(file))), FileSyncError,
storageWritePosixClose(file->driver), FileSyncError,
STORAGE_ERROR_WRITE_SYNC ": [9] Bad file descriptor", strZ(fileTmp));
// Disable file sync so close() can be reached
((StorageWritePosix *)ioWriteDriver(storageWriteIo(file)))->interface.syncFile = false;
((StorageWritePosix *)file->driver)->syncFile = false;
TEST_ERROR_FMT(
storageWritePosixClose(ioWriteDriver(storageWriteIo(file))), FileCloseError,
storageWritePosixClose(file->driver), FileCloseError,
STORAGE_ERROR_WRITE_CLOSE ": [9] Bad file descriptor", strZ(fileTmp));
// Set file descriptor to -1 so the close on free with not fail
((StorageWritePosix *)ioWriteDriver(storageWriteIo(file)))->fd = -1;
((StorageWritePosix *)file->driver)->fd = -1;
// -------------------------------------------------------------------------------------------------------------------------
TEST_TITLE("fail rename in close");
@@ -1401,7 +1401,7 @@ testRun(void)
strZ(fileTmp), strZ(fileName));
// Set file descriptor to -1 so the close on free with not fail
((StorageWritePosix *)ioWriteDriver(storageWriteIo(file)))->fd = -1;
((StorageWritePosix *)file->driver)->fd = -1;
storageRemoveP(storageTest, fileName, .errorOnMissing = true);
@@ -1732,8 +1732,7 @@ testRun(void)
StorageWrite *file = NULL;
TEST_ASSIGN(file, storageNewWriteP(storage, STRDEF("somefile"), .noSyncPath = false), "new file write");
TEST_RESULT_BOOL(
((StorageWritePosix *)ioWriteDriver(storageWriteIo(file)))->interface.syncPath, false, "path sync is disabled");
TEST_RESULT_BOOL(((StorageWritePosix *)file->driver)->syncPath, false, "path sync is disabled");
// -------------------------------------------------------------------------------------------------------------------------
TEST_TITLE("path sync result is noop");

6
test/src/module/storage/remoteTest.c
View File

@@ -451,9 +451,9 @@ testRun(void)
TEST_RESULT_VOID(storagePutP(write, contentBuf), "write file");
TEST_RESULT_UINT(
((StorageWriteRemote *)ioWriteDriver(storageWriteIo(write)))->protocolWriteBytes, bufSize(contentBuf),
((StorageWriteRemote *)write->driver)->protocolWriteBytes, bufSize(contentBuf),
"check write size");
TEST_RESULT_VOID(storageWriteRemoteClose(ioWriteDriver(storageWriteIo(write))), "close file again");
TEST_RESULT_VOID(storageWriteRemoteClose(write->driver), "close file again");
TEST_RESULT_VOID(storageWriteFree(write), "free file");
// Make sure the file was written correctly
@@ -509,7 +509,7 @@ testRun(void)
write, storageNewWriteP(storageRepoWrite, STRDEF("test2.txt"), .compressible = true), "new write file (compress)");
TEST_RESULT_VOID(storagePutP(write, contentBuf), "write file");
TEST_RESULT_BOOL(
((StorageWriteRemote *)ioWriteDriver(storageWriteIo(write)))->protocolWriteBytes < bufSize(contentBuf), true,
((StorageWriteRemote *)write->driver)->protocolWriteBytes < bufSize(contentBuf), true,
"check compressed write size");
}

9
test/src/module/storage/s3Test.c
View File

@@ -891,7 +891,7 @@ testRun(void)
TEST_RESULT_BOOL(storageWriteSyncPath(write), true, "path is synced");
TEST_RESULT_BOOL(storageWriteTruncate(write), true, "file will be truncated");
TEST_RESULT_VOID(storageWriteS3Close(ioWriteDriver(storageWriteIo(write))), "close file again");
TEST_RESULT_VOID(storageWriteS3Close(write->driver), "close file again");
// Check that temp credentials were changed
TEST_RESULT_STR_Z(driver->accessKey, "xx", "check access key");
@@ -1844,14 +1844,11 @@ testRun(void)
ioWriteOpen(storageWriteIo(write));
ioWrite(storageWriteIo(write), BUFSTRDEF("123456789012345678"));
TEST_RESULT_VOID(
bufResize(((StorageWriteS3 *)ioWriteDriver(storageWriteIo(write)))->partBuffer, 17),
"resize part buffer to 17");
TEST_RESULT_VOID(bufResize(((StorageWriteS3 *)write->driver)->partBuffer, 17), "resize part buffer to 17");
ioWrite(storageWriteIo(write), BUFSTRDEF("90"));
TEST_RESULT_UINT(
((StorageWriteS3 *)ioWriteDriver(storageWriteIo(write)))->partSize, 16, "part buffer reset to 16 (default)");
TEST_RESULT_UINT(((StorageWriteS3 *)write->driver)->partSize, 16, "part buffer reset to 16 (default)");
ioWriteClose(storageWriteIo(write));
ioBufferSizeSet(ioBufferSizeDefault);

2
test/src/module/storage/sftpTest.c
View File

@@ -5466,7 +5466,7 @@ testRun(void)
TEST_RESULT_INT(ioWriteFd(storageWriteIo(file)), -1, "check write fd");
// Make sftpHandle NULL
((StorageWriteSftp *)ioWriteDriver(storageWriteIo(file)))->sftpHandle = NULL;
((StorageWriteSftp *)file->driver)->sftpHandle = NULL;
TEST_RESULT_VOID(ioWriteClose(storageWriteIo(file)), "close file");