1

#include <stdlib.h>

     2

#include <assert.h>

     3

     4

#include <postgresql/libpq/libpq-fs.h>

     5

     6

#include "dbfs.h"

     7

#include "op_base.h"

     8

#include "../lib/log.h"

     9

    10

/*

    11

 * A file-operation, i.e. a sequence consisting of an OPEN, a multitude of READ/WRITE, followed by zero or more FLUSHes, and finally a single RELEASE.

    12

 *

    13

 * For historical reasons this opens a transaction and keeps it between open/release, but reads/writes now use the oid directly and are transactionless.

    14

 */

    15

    16

struct dbfs_fileop {

    17

    struct dbfs_op base;

    18

    19

    uint32_t oid;

    20

//    uint32_t lo_fd;

    21

};

    22

    23

static void _dbfs_fileop_free (struct dbfs_op *op_base) {

    24

    struct dbfs_fileop *fop = (struct dbfs_fileop *) op_base;

    25

    26

    /* no-op */

    27

    (void) fop;

    28

}

    29

    30

static void dbfs_open_res (struct evsql_result *res, void *arg) {

    31

    struct dbfs_fileop *fop = arg;

    32

    int err;

    33

    34

    // check the results

    35

    if ((err = _dbfs_check_res(res, 1, 2)))

    36

        SERROR(err = (err ==  1 ? ENOENT : EIO));

    37

    38

    const char *type;

    39

    40

    // extract the data

    41

    if (0

    42

        ||  evsql_result_string(res, 0, 0, &type,           0 ) // inodes.type

    43

        ||  evsql_result_uint32(res, 0, 1, &fop->oid,       0 ) // inodes.data

    44

    )

    45

        SERROR(err = EIO);

    46

    47

    // is it a dir?

    48

    if (_dbfs_mode(type) != S_IFREG)

    49

        EERROR(err = EINVAL, "wrong type: %s", type);

    50

    51

    INFO("\t[dbfs.open %p:%p] -> ino=%lu, type=%s", fop, fop->base.req, (unsigned long int) fop->base.ino, type);

    52

    53

    // open_fn done, do the open_reply

    54

    if ((err = dbfs_op_open_reply(&fop->base)))

    55

        goto error;

    56

    57

    // success, fallthrough for evsql_result_free

    58

    err = 0;

    59

    60

error:

    61

    if (err)

    62

        // fail it

    63

        dbfs_op_fail(&fop->base, err);

    64

    65

    // free

    66

    evsql_result_free(res);

    67

}

    68

    69

static void dbfs_fileop_open (struct dbfs_op *op_base) {

    70

    struct dbfs_fileop *fop = (struct dbfs_fileop *) op_base;

    71

    struct dbfs *ctx = fuse_req_userdata(fop->base.req);

    72

    int err;

    73

    74

    // make sure the file actually exists

    75

    const char *sql =

    76

        "SELECT"

    77

        " inodes.type, inodes.data"

    78

        " FROM inodes"

    79

        " WHERE inodes.ino = $1::int4";

    80

    81

    static struct evsql_query_params params = EVSQL_PARAMS(EVSQL_FMT_BINARY) {

    82

        EVSQL_PARAM ( UINT32 ),

    83

    84

        EVSQL_PARAMS_END

    85

    };

    86

    87

    // build params

    88

    if (0

    89

        ||  evsql_param_uint32(&params, 0, fop->base.ino)

    90

    )

    91

        SERROR(err = EIO);

    92

    93

    // query

    94

    if (evsql_query_params(ctx->db, fop->base.trans, sql, &params, dbfs_open_res, fop) == NULL)

    95

        SERROR(err = EIO);

    96

    97

    // ok, wait for the info results

    98

    return;

    99

   100

error:

   101

    // fail it

   102

    dbfs_op_fail(&fop->base, err);

   103

}

   104

   105

void dbfs_open (struct fuse_req *req, fuse_ino_t ino, struct fuse_file_info *fi) {

   106

    struct dbfs *ctx = fuse_req_userdata(req);

   107

    struct dbfs_fileop *fop = NULL;

   108

    int err;

   109

   110

    // allocate it

   111

    if ((fop = calloc(1, sizeof(*fop))) == NULL && (err = EIO))

   112

        ERROR("calloc");

   113

   114

    // do the op_open

   115

    if ((err = dbfs_op_open(ctx, &fop->base, req, ino, fi, _dbfs_fileop_free, dbfs_fileop_open)))

   116

        ERROR("dbfs_op_open");

   117

   118

    // log

   119

    INFO("[dbfs.open %p:%p] ino=%lu, fi->flags=%04X", fop, req, ino, fi->flags);

   120

   121

    // wait

   122

    return;

   123

   124

error:

   125

    if (fop) {

   126

        // we can fail normally

   127

        dbfs_op_fail(&fop->base, err);

   128

   129

    } else {

   130

        // must error out manually as we couldn't alloc the context

   131

        if ((err = -fuse_reply_err(req, err)))

   132

            EWARNING(err, "fuse_reply_err");

   133

    }

   134

}

   135

   136

void dbfs_read_res (struct evsql_result *res, void *arg) {

   137

    struct fuse_req *req = arg;

   138

    int err;

   139

    const char *buf;

   140

    size_t size;

   141

   142

    // check the results

   143

    if ((err = _dbfs_check_res(res, 1, 1)) < 0)

   144

        SERROR(err = EIO);

   145

   146

    // get the data

   147

    if (evsql_result_binary(res, 0, 0, &buf, &size, 0))

   148

        SERROR(err = EIO);

   149

   150

    INFO("\t[dbfs.read %p] -> size=%zu", req, size);

   151

   152

    // send it

   153

    if ((err = -fuse_reply_buf(req, buf, size)))

   154

        EERROR(err, "fuse_reply_buf");

   155

   156

    // good, fallthrough

   157

    err = 0;

   158

   159

error:

   160

    if (err)

   161

        fuse_reply_err(req, err);

   162

   163

   164

    // free

   165

    evsql_result_free(res);

   166

}

   167

   168

void dbfs_read (struct fuse_req *req, fuse_ino_t ino, size_t size, off_t off, struct fuse_file_info *fi) {

   169

    struct dbfs *ctx = fuse_req_userdata(req);

   170

    int err;

   171

   172

    // log

   173

    INFO("[dbfs.read %p] ino=%lu, size=%zu, off=%lu, fi->flags=%04X", req, ino, size, off, fi->flags);

   174

   175

    // query

   176

    const char *sql = 

   177

        "SELECT"

   178

        " lo_pread_oid(data, $1::int4, $2::int4)"

   179

        " FROM inodes"

   180

        " WHERE ino = $3::int4";

   181

   182

    static struct evsql_query_params params = EVSQL_PARAMS(EVSQL_FMT_BINARY) {

   183

        EVSQL_PARAM ( UINT32 ), // len

   184

        EVSQL_PARAM ( UINT32 ), // off

   185

        EVSQL_PARAM ( UINT32 ), // ino

   186

   187

        EVSQL_PARAMS_END

   188

    };

   189

   190

    // build params

   191

    if (0

   192

        ||  evsql_param_uint32(&params, 0, size)

   193

        ||  evsql_param_uint32(&params, 1, off)

   194

        ||  evsql_param_uint32(&params, 2, ino)

   195

    )

   196

        SERROR(err = EIO);

   197

   198

    // query, transactionless

   199

    if (evsql_query_params(ctx->db, NULL, sql, &params, dbfs_read_res, req) == NULL)

   200

        SERROR(err = EIO);

   201

   202

    // ok, wait for the info results

   203

    return;

   204

   205

error:

   206

    fuse_reply_err(req, err);

   207

}

   208

   209

void dbfs_write_res (struct evsql_result *res, void *arg) {

   210

    struct fuse_req *req = arg;

   211

    int err;

   212

    uint32_t size;

   213

   214

    // check the results

   215

    if ((err = _dbfs_check_res(res, 1, 1)) < 0)

   216

        SERROR(err = EIO);

   217

   218

    // get the size

   219

    if (evsql_result_uint32(res, 0, 0, &size, 0))

   220

        SERROR(err = EIO);

   221

   222

    INFO("\t[dbfs.write %p] -> size=%lu", req, (long unsigned int) size);

   223

   224

    // send it

   225

    if ((err = -fuse_reply_write(req, size)))

   226

        EERROR(err, "fuse_reply_write");

   227

   228

    // good, fallthrough

   229

    err = 0;

   230

   231

error:

   232

    if (err)

   233

        fuse_reply_err(req, err);

   234

   235

    // free

   236

    evsql_result_free(res);

   237

}

   238

   239

void dbfs_write (struct fuse_req *req, fuse_ino_t ino, const char *buf, size_t size, off_t off, struct fuse_file_info *fi) {

   240

    struct dbfs *ctx = fuse_req_userdata(req);

   241

    int err;

   242

   243

    // log

   244

    INFO("[dbfs.write %p] ino=%lu, size=%zu, off=%lu, fi->flags=%04X", req, ino, size, off, fi->flags);

   245

   246

    // query

   247

    const char *sql = 

   248

        "SELECT"

   249

        " lo_pwrite_oid(data, $1::bytea, $2::int4)"

   250

        " FROM inodes"

   251

        " WHERE ino = $3::int4";

   252

   253

    static struct evsql_query_params params = EVSQL_PARAMS(EVSQL_FMT_BINARY) {

   254

        EVSQL_PARAM ( BINARY ), // buf

   255

        EVSQL_PARAM ( UINT32 ), // off

   256

        EVSQL_PARAM ( UINT32 ), // oid

   257

   258

        EVSQL_PARAMS_END

   259

    };

   260

   261

    // build params

   262

    if (0

   263

        ||  evsql_param_binary(&params, 0, buf, size)

   264

        ||  evsql_param_uint32(&params, 1, off)

   265

        ||  evsql_param_uint32(&params, 2, ino)

   266

    )

   267

        SERROR(err = EIO);

   268

   269

    // query

   270

    if (evsql_query_params(ctx->db, NULL, sql, &params, dbfs_write_res, req) == NULL)

   271

        SERROR(err = EIO);

   272

   273

    // ok, wait for the info results

   274

    return;

   275

   276

error:

   277

    fuse_reply_err(req, err);

   278

}

   279

   280

void dbfs_flush (struct fuse_req *req, fuse_ino_t ino, struct fuse_file_info *fi) {

   281

    struct dbfs_fileop *fop;

   282

    int err;

   283

   284

    // get the fop

   285

    if ((fop = (struct dbfs_fileop *) dbfs_op_req(req, ino, fi)) == NULL)

   286

        return;

   287

   288

    // log

   289

    INFO("[dbfs.flush %p:%p] ino=%lu", fop, req, ino);

   290

   291

    // and reply...

   292

    if ((err = -fuse_reply_err(req, 0)))

   293

        EWARNING(err, "fuse_reply_err");

   294

   295

    // done

   296

    if ((err = dbfs_op_req_done(&fop->base)))

   297

        goto error;

   298

   299

    // good

   300

    return;

   301

   302

error:

   303

    dbfs_op_fail(&fop->base, err);

   304

}

   305

   306

void dbfs_release (struct fuse_req *req, fuse_ino_t ino, struct fuse_file_info *fi) {

   307

    // just passthrough to dbfs_op

   308

    // the lo_fd will be closed automatically

   309

    dbfs_op_release(req, ino, fi);

   310

}