Commit a634a113 authored by Kirill Smelkov's avatar Kirill Smelkov

.

parent 3c206aac
...@@ -466,7 +466,7 @@ error _Conn::resync(zodb::Tid at) { ...@@ -466,7 +466,7 @@ error _Conn::resync(zodb::Tid at) {
return err; // XXX -> wconn down on err ? return err; // XXX -> wconn down on err ?
// write-lock wconn._atMu . This excludes everything else, and in // lock wconn._atMu.W . This excludes everything else, and in
// particular _pinner_, from running and mutating files and mappings. // particular _pinner_, from running and mutating files and mappings.
// //
// NOTE we'll relock atMu as R in the second part of resync, so we prelock // NOTE we'll relock atMu as R in the second part of resync, so we prelock
...@@ -530,6 +530,8 @@ error _Conn::resync(zodb::Tid at) { ...@@ -530,6 +530,8 @@ error _Conn::resync(zodb::Tid at) {
if (!(headfsize % f->blksize == 0)) if (!(headfsize % f->blksize == 0))
return E(fmt::errorf("wcfs bug: head/file size %% blksize != 0")); return E(fmt::errorf("wcfs bug: head/file size %% blksize != 0"));
// replace zero regions in f mappings in accordance to adjusted f._headfsize.
// NOTE it is ok to access f._mmaps without locking f._mu because we hold wconn.atMu.W
for (auto mmap : f->_mmaps) { for (auto mmap : f->_mmaps) {
//printf(" resync -> %s: unzero [%lu:%lu)", v(at), f->_headfsize/f->blksize, headfsize/f->blksize); //printf(" resync -> %s: unzero [%lu:%lu)", v(at), f->_headfsize/f->blksize, headfsize/f->blksize);
uint8_t *mem_unzero_start = min(mmap->mem_stop, uint8_t *mem_unzero_start = min(mmap->mem_stop,
...@@ -571,6 +573,7 @@ error _Conn::resync(zodb::Tid at) { ...@@ -571,6 +573,7 @@ error _Conn::resync(zodb::Tid at) {
zodb::Oid foid = fit.first; zodb::Oid foid = fit.first;
//FileH f = fit.second; //FileH f = fit.second;
// XXX need to lock f.mu because wconn.atMu is only R now.
string ack; string ack;
tie(ack, err) = wconn._wlink->sendReq(context::background(), fmt::sprintf("watch %s @%s", v(foid), v(at))); tie(ack, err) = wconn._wlink->sendReq(context::background(), fmt::sprintf("watch %s @%s", v(foid), v(at)));
if (err != nil) if (err != nil)
...@@ -702,37 +705,54 @@ error _FileH::close() { ...@@ -702,37 +705,54 @@ error _FileH::close() {
_FileH& fileh = *this; _FileH& fileh = *this;
Conn wconn = fileh.wconn; Conn wconn = fileh.wconn;
// XXX decref open count
// XXX fileh.close can be called several times and after first call another
// XXX fileh could be opened for the same foid. Be careful not to interfere with it.
// -> just return nil for second close
// XXX locking ok? // XXX locking ok?
wconn->_atMu.RLock(); wconn->_atMu.RLock();
fileh._mu.lock();
defer([&]() { defer([&]() {
fileh._mu.unlock();
wconn->_atMu.RUnlock(); wconn->_atMu.RUnlock();
}); });
// fileh.close can be called several times. just return nil for second close.
if (fileh._closed)
return nil;
// decref open count; do real close only when last open goes away.
if (fileh._nopen <= 0)
panic("BUG: fileh._nopen <= 0");
fileh._nopen--;
if (fileh._nopen > 0)
return nil;
// last open went away - real close.
xerr::Contextf E("%s: close f<%s>", v(wconn), v(fileh.foid)); xerr::Contextf E("%s: close f<%s>", v(wconn), v(fileh.foid));
error err, eret;
auto reterr1 = [&eret](error err) {
if (eret == nil && err != nil)
eret = err;
};
// XXX change all fileh.mmaps to cause EFAULT on any access after fileh.close // XXX change all fileh.mmaps to cause EFAULT on any access after fileh.close
// stop watching f // stop watching f XXX ok under f.mu ?
string ack; string ack;
error err;
tie(ack, err) = wconn->_wlink->sendReq(context::background(), fmt::sprintf("watch %s -", v(foid))); tie(ack, err) = wconn->_wlink->sendReq(context::background(), fmt::sprintf("watch %s -", v(foid)));
if (err != nil) if (err != nil)
return E(err); reterr1(err);
if (ack != "ok") else if (ack != "ok")
return E(fmt::errorf("unwatch: %s", v(ack))); reterr1(fmt::errorf("unwatch: %s", v(ack)));
// remove fileh from wconn._filehTab // remove fileh from wconn._filehTab
wconn->_mu.Lock(); wconn->_mu.Lock(); // FIXME lock order vs fileh._mu
if (wconn->_filehTab.get(fileh.foid)._ptr() == &fileh) // XXX -> assert ? if (wconn->_filehTab.get(fileh.foid)._ptr() == &fileh) // XXX -> panic(BUG)
wconn->_filehTab.erase(fileh.foid); wconn->_filehTab.erase(fileh.foid);
wconn->_mu.Unlock(); wconn->_mu.Unlock();
return E(fileh._headf->close()); reterr1(fileh._headf->close());
fileh._closed = true;
return E(eret);
} }
// mmap creates file mapping representing file[blk_start +blk_len) data as of wconn.at database state. // mmap creates file mapping representing file[blk_start +blk_len) data as of wconn.at database state.
...@@ -863,6 +883,7 @@ error _Mapping::unmap() { ...@@ -863,6 +883,7 @@ error _Mapping::unmap() {
// XXX clear other fields? // XXX clear other fields?
// XXX do it first? (to avoid pinner going through f.mmaps and hitting unmapped memory) // XXX do it first? (to avoid pinner going through f.mmaps and hitting unmapped memory)
// -> no need: both pinner and unmap lock on f.mu
//f->_mmaps.remove(mmap); //f->_mmaps.remove(mmap);
f->_mmaps.erase( f->_mmaps.erase(
std::remove(f->_mmaps.begin(), f->_mmaps.end(), mmap), std::remove(f->_mmaps.begin(), f->_mmaps.end(), mmap),
......
...@@ -238,6 +238,8 @@ struct _FileH : object { ...@@ -238,6 +238,8 @@ struct _FileH : object {
sync::Mutex _mu; // atMu.W | atMu.R + _mu sync::Mutex _mu; // atMu.W | atMu.R + _mu
dict<int64_t, zodb::Tid> _pinned; // {} blk -> rev that wcfs already sent us for this file dict<int64_t, zodb::Tid> _pinned; // {} blk -> rev that wcfs already sent us for this file
vector<Mapping> _mmaps; // []Mapping ↑blk_start mappings of this file vector<Mapping> _mmaps; // []Mapping ↑blk_start mappings of this file
int _nopen; // number of times Conn.open returned this fileh
bool _closed; // y after .close()
// don't new - create via Conn.open // don't new - create via Conn.open
private: private:
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment