Begin journal flusher
parent
5ac3910cef
commit
f892104aed
|
@ -14,6 +14,7 @@
|
||||||
#include <vector>
|
#include <vector>
|
||||||
#include <map>
|
#include <map>
|
||||||
#include <list>
|
#include <list>
|
||||||
|
#include <deque>
|
||||||
#include <set>
|
#include <set>
|
||||||
#include <functional>
|
#include <functional>
|
||||||
|
|
||||||
|
@ -143,7 +144,7 @@ struct __attribute__((__packed__)) dirty_entry
|
||||||
uint32_t flags;
|
uint32_t flags;
|
||||||
uint64_t location; // location in either journal or data
|
uint64_t location; // location in either journal or data
|
||||||
uint32_t offset; // offset within stripe
|
uint32_t offset; // offset within stripe
|
||||||
uint32_t size; // entry size
|
uint32_t size; // entry size. FIXME: rename to len?
|
||||||
};
|
};
|
||||||
|
|
||||||
class oid_hash
|
class oid_hash
|
||||||
|
@ -297,10 +298,11 @@ class blockstore
|
||||||
int continue_sync(blockstore_operation *op);
|
int continue_sync(blockstore_operation *op);
|
||||||
int ack_sync(blockstore_operation *op);
|
int ack_sync(blockstore_operation *op);
|
||||||
|
|
||||||
// Stable
|
// Stabilize
|
||||||
int dequeue_stable(blockstore_operation *op);
|
int dequeue_stable(blockstore_operation *op);
|
||||||
int continue_stable(blockstore_operation *op);
|
int continue_stable(blockstore_operation *op);
|
||||||
void handle_stable_event(ring_data_t *data, blockstore_operation *op);
|
void handle_stable_event(ring_data_t *data, blockstore_operation *op);
|
||||||
|
void stabilize_object(object_id oid, uint64_t max_ver);
|
||||||
|
|
||||||
public:
|
public:
|
||||||
|
|
||||||
|
|
|
@ -23,6 +23,7 @@ int blockstore::fulfill_read_push(blockstore_operation *op, uint64_t &fulfilled,
|
||||||
op->buf + cur_start - op->offset,
|
op->buf + cur_start - op->offset,
|
||||||
cur_end - cur_start
|
cur_end - cur_start
|
||||||
};
|
};
|
||||||
|
// FIXME: use simple std::vector instead of map for read_vec
|
||||||
op->read_vec[cur_start] = data->iov;
|
op->read_vec[cur_start] = data->iov;
|
||||||
io_uring_prep_readv(
|
io_uring_prep_readv(
|
||||||
sqe,
|
sqe,
|
||||||
|
|
|
@ -149,9 +149,74 @@ void blockstore::handle_stable_event(ring_data_t *data, blockstore_operation *op
|
||||||
dirty_it--;
|
dirty_it--;
|
||||||
} while (dirty_it != dirty_db.begin() && dirty_it->first.oid == v->oid);
|
} while (dirty_it != dirty_db.begin() && dirty_it->first.oid == v->oid);
|
||||||
}
|
}
|
||||||
// Acknowledge op
|
|
||||||
op->retval = 0;
|
|
||||||
op->callback(op);
|
|
||||||
}
|
}
|
||||||
|
// Acknowledge op
|
||||||
|
op->retval = 0;
|
||||||
|
op->callback(op);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
struct offset_len
|
||||||
|
{
|
||||||
|
uint64_t offset, len;
|
||||||
|
};
|
||||||
|
|
||||||
|
struct journal_flusher_t
|
||||||
|
{
|
||||||
|
std::deque<obj_ver_id> flush_queue;
|
||||||
|
obj_ver_id cur;
|
||||||
|
std::map<obj_ver_id, dirty_entry>::iterator dirty_it;
|
||||||
|
std::vector<offset_len> v;
|
||||||
|
};
|
||||||
|
|
||||||
|
void blockstore::stabilize_object(object_id oid, uint64_t max_ver)
|
||||||
|
{
|
||||||
|
auto dirty_it = dirty_db.find((obj_ver_id){
|
||||||
|
.oid = oid,
|
||||||
|
.version = max_ver,
|
||||||
|
});
|
||||||
|
if (dirty_it != dirty_db.end())
|
||||||
|
{
|
||||||
|
std::vector<offset_len> v;
|
||||||
|
do
|
||||||
|
{
|
||||||
|
if (dirty_it->second.state == ST_J_STABLE)
|
||||||
|
{
|
||||||
|
uint64_t offset = dirty_it->second.offset, len = dirty_it->second.size;
|
||||||
|
auto it = v.begin();
|
||||||
|
while (1)
|
||||||
|
{
|
||||||
|
for (; it != v.end(); it++)
|
||||||
|
if (it->offset >= offset)
|
||||||
|
break;
|
||||||
|
if (it == v.end() || it->offset >= offset+len)
|
||||||
|
{
|
||||||
|
v.insert(it, (offset_len){ .offset = offset, .len = len });
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
else
|
||||||
|
{
|
||||||
|
if (it->offset > offset)
|
||||||
|
v.insert(it, (offset_len){ .offset = offset, .len = it->offset-offset });
|
||||||
|
if (offset+len > it->offset+it->len)
|
||||||
|
{
|
||||||
|
len = offset+len - (it->offset+it->len);
|
||||||
|
offset = it->offset+it->len;
|
||||||
|
}
|
||||||
|
else
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
else if (dirty_it->second.state == ST_D_STABLE)
|
||||||
|
{
|
||||||
|
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
else if (IS_STABLE(dirty_it->second.state))
|
||||||
|
{
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
} while (dirty_it != dirty_db.begin());
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
Loading…
Reference in New Issue