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

struct dbfs_fileop {

    11

    struct dbfs_op base;

    12

    13

    uint32_t lo_fd;

    14

};

    15

    16

static void _dbfs_fileop_free (struct dbfs_op *op_base) {

    17

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

    18

    19

    /* no-op */

    20

    (void) fop;

    21

}

    22

    23

static void dbfs_open_res (const struct evsql_result_info *res, void *arg) {

    24

    struct dbfs_fileop *fop = arg;

    25

    int err;

    26

    27

    // check the results

    28

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

    29

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

    30

    31

    const char *type;

    32

    33

    // extract the data

    34

    if (0

    35

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

    36

        ||  evsql_result_uint32(res, 0, 1, &fop->lo_fd,     0 ) // fd

    37

    )

    38

        SERROR(err = EIO);

    39

    40

    // is it a dir?

    41

    if (_dbfs_mode(type) != S_IFREG)

    42

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

    43

    45

    46

    // open_fn done, do the open_reply

    47

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

    48

        goto error;

    49

    50

    // success, fallthrough for evsql_result_free

    51

    err = 0;

    52

    53

error:

    54

    if (err)

    55

        // fail it

    56

        dbfs_op_fail(&fop->base, err);

    57

    58

    // free

    59

    evsql_result_free(res);

    60

}

    61

    62

static void dbfs_fileop_open (struct dbfs_op *op_base) {

    63

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

    64

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

    65

    int err;

    66

    67

    // make sure the file actually exists

    68

    const char *sql =

    69

        "SELECT"

    70

        " inodes.type, lo_open(inodes.data, $1::int4) AS fd"

    71

        " FROM inodes"

    72

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

    73

    74

    static struct evsql_query_params params = EVSQL_PARAMS(EVSQL_FMT_BINARY) {

    75

        EVSQL_PARAM ( UINT32 ),

    76

        EVSQL_PARAM ( UINT32 ),

    77

    78

        EVSQL_PARAMS_END

    79

    };

    80

    81

    // build params

    82

    if (0

    83

        ||  evsql_param_uint32(&params, 0, INV_READ | INV_WRITE)

    84

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

    85

    )

    86

        SERROR(err = EIO);

    87

    88

    // query

    89

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

    90

        SERROR(err = EIO);

    91

    92

    // ok, wait for the info results

    93

    return;

    94

    95

error:

    96

    // fail it

    97

    dbfs_op_fail(&fop->base, err);

    98

}

    99

   100

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

   101

    struct dbfs *ctx = fuse_req_userdata(req);

   102

    struct dbfs_fileop *fop = NULL;

   103

    int err;

   104

   105

    // allocate it

   106

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

   107

        ERROR("calloc");

   108

   109

    // do the op_open

   110

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

   111

        ERROR("dbfs_op_open");

   112

   113

    // log

   114

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

   115

   116

    // wait

   117

    return;

   118

   119

error:

   120

    if (fop) {

   121

        // we can fail normally

   122

        dbfs_op_fail(&fop->base, err);

   123

   124

    } else {

   125

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

   126

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

   127

            EWARNING(err, "fuse_reply_err");

   128

    }

   129

}

   130

   131

void dbfs_read_res (const struct evsql_result_info *res, void *arg) {

   132

    struct dbfs_fileop *fop = arg;

   133

    int err;

   134

    const char *buf;

   135

    size_t size;

   136

   137

    // check the results

   138

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

   139

        SERROR(err = EIO);

   140

   141

    // get the data

   142

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

   143

        SERROR(err = EIO);

   144

   146

   147

    // send it

   148

    if ((err = fuse_reply_buf(fop->base.req, buf, size)))

   149

        EERROR(err, "fuse_reply_buf");

   150

   151

    // ok, req handled

   152

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

   153

        goto error;

   154

   155

    // good, fallthrough

   156

    err = 0;

   157

   158

error:

   159

    if (err)

   160

        dbfs_op_fail(&fop->base, err);

   161

   162

    // free

   163

    evsql_result_free(res);

   164

}

   165

   166

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

   167

    struct dbfs *ctx = fuse_req_userdata(req);

   168

    struct dbfs_fileop *fop;

   169

    int err;

   170

   171

    // get the op

   172

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

   173

        return;

   174

   175

    // log

   176

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

   177

   178

    // query

   179

    const char *sql = 

   180

        "SELECT"

   181

        " lo_pread($1::int4, $2::int4, $3::int4)";

   182

   183

    static struct evsql_query_params params = EVSQL_PARAMS(EVSQL_FMT_BINARY) {

   184

        EVSQL_PARAM ( UINT32 ), // fd

   185

        EVSQL_PARAM ( UINT32 ), // len

   186

        EVSQL_PARAM ( UINT32 ), // off

   187

   188

        EVSQL_PARAMS_END

   189

    };

   190

   191

    // build params

   192

    if (0

   193

        ||  evsql_param_uint32(&params, 0, fop->lo_fd)

   194

        ||  evsql_param_uint32(&params, 1, size)

   195

        ||  evsql_param_uint32(&params, 2, off)

   196

    )

   197

        SERROR(err = EIO);

   198

   199

    // query

   200

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

   201

        SERROR(err = EIO);

   202

   203

    // ok, wait for the info results

   204

    return;

   205

   206

error:

   207

    // fail it

   208

    dbfs_op_fail(&fop->base, err);

   209

}

   210

   211

void dbfs_write_res (const struct evsql_result_info *res, void *arg) {

   212

    struct dbfs_fileop *fop = arg;

   213

    int err;

   214

    uint32_t size;

   215

   216

    // check the results

   217

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

   218

        SERROR(err = EIO);

   219

   220

    // get the size

   221

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

   222

        SERROR(err = EIO);

   223

   225

   226

    // send it

   227

    if ((err = fuse_reply_write(fop->base.req, size)))

   228

        EERROR(err, "fuse_reply_write");

   229

   230

    // ok, req handled

   231

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

   232

        goto error;

   233

   234

    // good, fallthrough

   235

    err = 0;

   236

   237

error:

   238

    if (err)

   239

        dbfs_op_fail(&fop->base, err);

   240

   241

    // free

   242

    evsql_result_free(res);

   243

}

   244

   245

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) {

   246

    struct dbfs *ctx = fuse_req_userdata(req);

   247

    struct dbfs_fileop *fop;

   248

    int err;

   249

   250

    // get the op

   251

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

   252

        return;

   253

   254

    // log

   255

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

   256

   257

    // query

   258

    const char *sql = 

   259

        "SELECT"

   260

        " lo_pwrite($1::int4, $2::bytea, $3::int4)";

   261

   262

    static struct evsql_query_params params = EVSQL_PARAMS(EVSQL_FMT_BINARY) {

   263

        EVSQL_PARAM ( UINT32 ), // fd

   264

        EVSQL_PARAM ( BINARY ), // buf

   265

        EVSQL_PARAM ( UINT32 ), // off

   266

   267

        EVSQL_PARAMS_END

   268

    };

   269

   270

    // build params

   271

    if (0

   272

        ||  evsql_param_uint32(&params, 0, fop->lo_fd)

   273

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

   274

        ||  evsql_param_uint32(&params, 2, off)

   275

    )

   276

        SERROR(err = EIO);

   277

   278

    // query

   279

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

   280

        SERROR(err = EIO);

   281

   282

    // ok, wait for the info results

   283

    return;

   284

   285

error:

   286

    // fail it

   287

    dbfs_op_fail(&fop->base, err);

   288

}

   289

   290

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

   291

    struct dbfs_fileop *fop;

   292

    int err;

   293

   294

    // get the fop

   295

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

   296

        return;

   297

   298

    // log

   299

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

   300

   301

    // and reply...

   302

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

   303

        EWARNING(err, "fuse_reply_err");

   304

   305

    // done

   306

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

   307

        goto error;

   308

   309

    // good

   310

    return;

   311

   312

error:

   313

    dbfs_op_fail(&fop->base, err);

   314

}

   315

   316

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

   317

    // just passthrough to dbfs_op

   318

    // the lo_fd will be closed automatically

   319

    dbfs_op_release(req, ino, fi);

   320

}