--- /dev/null
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2004-2006 Sage Weil <sage@newdream.net>
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software
+ * Foundation. See file COPYING.
+ *
+ */
+
+
+#include <mutex>
+#include <algorithm>
+#include "Filer.h"
+#include "osd/OSDMap.h"
+#include "Striper.h"
+
+#include "messages/MOSDOp.h"
+#include "messages/MOSDOpReply.h"
+#include "messages/MOSDMap.h"
+
+#include "msg/Messenger.h"
+
+#include "include/Context.h"
+
+#include "common/Finisher.h"
+#include "common/config.h"
+
+#define dout_subsys ceph_subsys_filer
+#undef dout_prefix
+#define dout_prefix *_dout << objecter->messenger->get_myname() << ".filer "
+
+class Filer::C_Probe : public Context {
+public:
+ Filer *filer;
+ Probe *probe;
+ object_t oid;
+ uint64_t size;
+ ceph::real_time mtime;
+ C_Probe(Filer *f, Probe *p, object_t o) : filer(f), probe(p), oid(o),
+ size(0) {}
+ void finish(int r) override {
+ if (r == -ENOENT) {
+ r = 0;
+ assert(size == 0);
+ }
+
+ bool probe_complete;
+ {
+ Probe::unique_lock pl(probe->lock);
+ if (r != 0) {
+ probe->err = r;
+ }
+
+ probe_complete = filer->_probed(probe, oid, size, mtime, pl);
+ assert(!pl.owns_lock());
+ }
+ if (probe_complete) {
+ probe->onfinish->complete(probe->err);
+ delete probe;
+ }
+ }
+};
+
+int Filer::probe(inodeno_t ino,
+ file_layout_t *layout,
+ snapid_t snapid,
+ uint64_t start_from,
+ uint64_t *end, // LB, when !fwd
+ ceph::real_time *pmtime,
+ bool fwd,
+ int flags,
+ Context *onfinish)
+{
+ ldout(cct, 10) << "probe " << (fwd ? "fwd ":"bwd ")
+ << hex << ino << dec
+ << " starting from " << start_from
+ << dendl;
+
+ assert(snapid); // (until there is a non-NOSNAP write)
+
+ Probe *probe = new Probe(ino, *layout, snapid, start_from, end, pmtime,
+ flags, fwd, onfinish);
+
+ return probe_impl(probe, layout, start_from, end);
+}
+
+int Filer::probe(inodeno_t ino,
+ file_layout_t *layout,
+ snapid_t snapid,
+ uint64_t start_from,
+ uint64_t *end, // LB, when !fwd
+ utime_t *pmtime,
+ bool fwd,
+ int flags,
+ Context *onfinish)
+{
+ ldout(cct, 10) << "probe " << (fwd ? "fwd ":"bwd ")
+ << hex << ino << dec
+ << " starting from " << start_from
+ << dendl;
+
+ assert(snapid); // (until there is a non-NOSNAP write)
+
+ Probe *probe = new Probe(ino, *layout, snapid, start_from, end, pmtime,
+ flags, fwd, onfinish);
+ return probe_impl(probe, layout, start_from, end);
+}
+
+int Filer::probe_impl(Probe* probe, file_layout_t *layout,
+ uint64_t start_from, uint64_t *end) // LB, when !fwd
+{
+ // period (bytes before we jump unto a new set of object(s))
+ uint64_t period = layout->get_period();
+
+ // start with 1+ periods.
+ probe->probing_len = period;
+ if (probe->fwd) {
+ if (start_from % period)
+ probe->probing_len += period - (start_from % period);
+ } else {
+ assert(start_from > *end);
+ if (start_from % period)
+ probe->probing_len -= period - (start_from % period);
+ probe->probing_off -= probe->probing_len;
+ }
+
+ Probe::unique_lock pl(probe->lock);
+ _probe(probe, pl);
+ assert(!pl.owns_lock());
+
+ return 0;
+}
+
+
+
+/**
+ * probe->lock must be initially locked, this function will release it
+ */
+void Filer::_probe(Probe *probe, Probe::unique_lock& pl)
+{
+ assert(pl.owns_lock() && pl.mutex() == &probe->lock);
+
+ ldout(cct, 10) << "_probe " << hex << probe->ino << dec
+ << " " << probe->probing_off << "~" << probe->probing_len
+ << dendl;
+
+ // map range onto objects
+ probe->known_size.clear();
+ probe->probing.clear();
+ Striper::file_to_extents(cct, probe->ino, &probe->layout, probe->probing_off,
+ probe->probing_len, 0, probe->probing);
+
+ std::vector<ObjectExtent> stat_extents;
+ for (vector<ObjectExtent>::iterator p = probe->probing.begin();
+ p != probe->probing.end();
+ ++p) {
+ ldout(cct, 10) << "_probe probing " << p->oid << dendl;
+ probe->ops.insert(p->oid);
+ stat_extents.push_back(*p);
+ }
+
+ pl.unlock();
+ for (std::vector<ObjectExtent>::iterator i = stat_extents.begin();
+ i != stat_extents.end(); ++i) {
+ C_Probe *c = new C_Probe(this, probe, i->oid);
+ objecter->stat(i->oid, i->oloc, probe->snapid, &c->size, &c->mtime,
+ probe->flags | CEPH_OSD_FLAG_RWORDERED,
+ new C_OnFinisher(c, finisher));
+ }
+}
+
+/**
+ * probe->lock must be initially held, and will be released by this function.
+ *
+ * @return true if probe is complete and Probe object may be freed.
+ */
+bool Filer::_probed(Probe *probe, const object_t& oid, uint64_t size,
+ ceph::real_time mtime, Probe::unique_lock& pl)
+{
+ assert(pl.owns_lock() && pl.mutex() == &probe->lock);
+
+ ldout(cct, 10) << "_probed " << probe->ino << " object " << oid
+ << " has size " << size << " mtime " << mtime << dendl;
+
+ probe->known_size[oid] = size;
+ if (mtime > probe->max_mtime)
+ probe->max_mtime = mtime;
+
+ assert(probe->ops.count(oid));
+ probe->ops.erase(oid);
+
+ if (!probe->ops.empty()) {
+ pl.unlock();
+ return false; // waiting for more!
+ }
+
+ if (probe->err) { // we hit an error, propagate back up
+ pl.unlock();
+ return true;
+ }
+
+ // analyze!
+ uint64_t end = 0;
+
+ if (!probe->fwd) {
+ std::reverse(probe->probing.begin(), probe->probing.end());
+ }
+
+ for (vector<ObjectExtent>::iterator p = probe->probing.begin();
+ p != probe->probing.end();
+ ++p) {
+ uint64_t shouldbe = p->length + p->offset;
+ ldout(cct, 10) << "_probed " << probe->ino << " object " << hex
+ << p->oid << dec << " should be " << shouldbe
+ << ", actual is " << probe->known_size[p->oid]
+ << dendl;
+
+ if (!probe->found_size) {
+ assert(probe->known_size[p->oid] <= shouldbe);
+
+ if ((probe->fwd && probe->known_size[p->oid] == shouldbe) ||
+ (!probe->fwd && probe->known_size[p->oid] == 0 &&
+ probe->probing_off > 0))
+ continue; // keep going
+
+ // aha, we found the end!
+ // calc offset into buffer_extent to get distance from probe->from.
+ uint64_t oleft = probe->known_size[p->oid] - p->offset;
+ for (vector<pair<uint64_t, uint64_t> >::iterator i
+ = p->buffer_extents.begin();
+ i != p->buffer_extents.end();
+ ++i) {
+ if (oleft <= (uint64_t)i->second) {
+ end = probe->probing_off + i->first + oleft;
+ ldout(cct, 10) << "_probed end is in buffer_extent " << i->first
+ << "~" << i->second << " off " << oleft
+ << ", from was " << probe->probing_off << ", end is "
+ << end << dendl;
+
+ probe->found_size = true;
+ ldout(cct, 10) << "_probed found size at " << end << dendl;
+ *probe->psize = end;
+
+ if (!probe->pmtime &&
+ !probe->pumtime) // stop if we don't need mtime too
+ break;
+ }
+ oleft -= i->second;
+ }
+ }
+ break;
+ }
+
+ if (!probe->found_size || (probe->probing_off && (probe->pmtime ||
+ probe->pumtime))) {
+ // keep probing!
+ ldout(cct, 10) << "_probed probing further" << dendl;
+
+ uint64_t period = probe->layout.get_period();
+ if (probe->fwd) {
+ probe->probing_off += probe->probing_len;
+ assert(probe->probing_off % period == 0);
+ probe->probing_len = period;
+ } else {
+ // previous period.
+ assert(probe->probing_off % period == 0);
+ probe->probing_len = period;
+ probe->probing_off -= period;
+ }
+ _probe(probe, pl);
+ assert(!pl.owns_lock());
+ return false;
+ } else if (probe->pmtime) {
+ ldout(cct, 10) << "_probed found mtime " << probe->max_mtime << dendl;
+ *probe->pmtime = probe->max_mtime;
+ } else if (probe->pumtime) {
+ ldout(cct, 10) << "_probed found mtime " << probe->max_mtime << dendl;
+ *probe->pumtime = ceph::real_clock::to_ceph_timespec(probe->max_mtime);
+ }
+ // done!
+ pl.unlock();
+ return true;
+}
+
+
+// -----------------------
+
+struct PurgeRange {
+ std::mutex lock;
+ typedef std::lock_guard<std::mutex> lock_guard;
+ typedef std::unique_lock<std::mutex> unique_lock;
+ inodeno_t ino;
+ file_layout_t layout;
+ SnapContext snapc;
+ uint64_t first, num;
+ ceph::real_time mtime;
+ int flags;
+ Context *oncommit;
+ int uncommitted;
+ PurgeRange(inodeno_t i, const file_layout_t& l, const SnapContext& sc,
+ uint64_t fo, uint64_t no, ceph::real_time t, int fl,
+ Context *fin)
+ : ino(i), layout(l), snapc(sc), first(fo), num(no), mtime(t), flags(fl),
+ oncommit(fin), uncommitted(0) {}
+};
+
+int Filer::purge_range(inodeno_t ino,
+ const file_layout_t *layout,
+ const SnapContext& snapc,
+ uint64_t first_obj, uint64_t num_obj,
+ ceph::real_time mtime,
+ int flags,
+ Context *oncommit)
+{
+ assert(num_obj > 0);
+
+ // single object? easy!
+ if (num_obj == 1) {
+ object_t oid = file_object_t(ino, first_obj);
+ object_locator_t oloc = OSDMap::file_to_object_locator(*layout);
+ objecter->remove(oid, oloc, snapc, mtime, flags, oncommit);
+ return 0;
+ }
+
+ PurgeRange *pr = new PurgeRange(ino, *layout, snapc, first_obj,
+ num_obj, mtime, flags, oncommit);
+
+ _do_purge_range(pr, 0);
+ return 0;
+}
+
+struct C_PurgeRange : public Context {
+ Filer *filer;
+ PurgeRange *pr;
+ C_PurgeRange(Filer *f, PurgeRange *p) : filer(f), pr(p) {}
+ void finish(int r) override {
+ filer->_do_purge_range(pr, 1);
+ }
+};
+
+void Filer::_do_purge_range(PurgeRange *pr, int fin)
+{
+ PurgeRange::unique_lock prl(pr->lock);
+ pr->uncommitted -= fin;
+ ldout(cct, 10) << "_do_purge_range " << pr->ino << " objects " << pr->first
+ << "~" << pr->num << " uncommitted " << pr->uncommitted
+ << dendl;
+
+ if (pr->num == 0 && pr->uncommitted == 0) {
+ pr->oncommit->complete(0);
+ prl.unlock();
+ delete pr;
+ return;
+ }
+
+ std::vector<object_t> remove_oids;
+
+ int max = cct->_conf->filer_max_purge_ops - pr->uncommitted;
+ while (pr->num > 0 && max > 0) {
+ remove_oids.push_back(file_object_t(pr->ino, pr->first));
+ pr->uncommitted++;
+ pr->first++;
+ pr->num--;
+ max--;
+ }
+ prl.unlock();
+
+ // Issue objecter ops outside pr->lock to avoid lock dependency loop
+ for (const auto& oid : remove_oids) {
+ object_locator_t oloc = OSDMap::file_to_object_locator(pr->layout);
+ objecter->remove(oid, oloc, pr->snapc, pr->mtime, pr->flags,
+ new C_OnFinisher(new C_PurgeRange(this, pr), finisher));
+ }
+}
+
+// -----------------------
+struct TruncRange {
+ std::mutex lock;
+ typedef std::lock_guard<std::mutex> lock_guard;
+ typedef std::unique_lock<std::mutex> unique_lock;
+ inodeno_t ino;
+ file_layout_t layout;
+ SnapContext snapc;
+ ceph::real_time mtime;
+ int flags;
+ Context *oncommit;
+ int uncommitted;
+ uint64_t offset;
+ uint64_t length;
+ uint32_t truncate_seq;
+ TruncRange(inodeno_t i, const file_layout_t& l, const SnapContext& sc,
+ ceph::real_time t, int fl, Context *fin,
+ uint64_t off, uint64_t len, uint32_t ts)
+ : ino(i), layout(l), snapc(sc), mtime(t), flags(fl), oncommit(fin),
+ uncommitted(0), offset(off), length(len), truncate_seq(ts) {}
+};
+
+void Filer::truncate(inodeno_t ino,
+ file_layout_t *layout,
+ const SnapContext& snapc,
+ uint64_t offset,
+ uint64_t len,
+ __u32 truncate_seq,
+ ceph::real_time mtime,
+ int flags,
+ Context *oncommit)
+{
+ uint64_t period = layout->get_period();
+ uint64_t num_objs = Striper::get_num_objects(*layout, len + (offset % period));
+ if (num_objs == 1) {
+ vector<ObjectExtent> extents;
+ Striper::file_to_extents(cct, ino, layout, offset, len, 0, extents);
+ vector<OSDOp> ops(1);
+ ops[0].op.op = CEPH_OSD_OP_TRIMTRUNC;
+ ops[0].op.extent.truncate_seq = truncate_seq;
+ ops[0].op.extent.truncate_size = extents[0].offset;
+ objecter->_modify(extents[0].oid, extents[0].oloc, ops, mtime, snapc,
+ flags, oncommit);
+ return;
+ }
+
+ if (len > 0 && (offset + len) % period)
+ len += period - ((offset + len) % period);
+
+ TruncRange *tr = new TruncRange(ino, *layout, snapc, mtime, flags, oncommit,
+ offset, len, truncate_seq);
+ _do_truncate_range(tr, 0);
+}
+
+struct C_TruncRange : public Context {
+ Filer *filer;
+ TruncRange *tr;
+ C_TruncRange(Filer *f, TruncRange *t) : filer(f), tr(t) {}
+ void finish(int r) override {
+ filer->_do_truncate_range(tr, 1);
+ }
+};
+
+void Filer::_do_truncate_range(TruncRange *tr, int fin)
+{
+ TruncRange::unique_lock trl(tr->lock);
+ tr->uncommitted -= fin;
+ ldout(cct, 10) << "_do_truncate_range " << tr->ino << " objects " << tr->offset
+ << "~" << tr->length << " uncommitted " << tr->uncommitted
+ << dendl;
+
+ if (tr->length == 0 && tr->uncommitted == 0) {
+ tr->oncommit->complete(0);
+ trl.unlock();
+ delete tr;
+ return;
+ }
+
+ vector<ObjectExtent> extents;
+
+ int max = cct->_conf->filer_max_truncate_ops - tr->uncommitted;
+ if (max > 0 && tr->length > 0) {
+ uint64_t len = tr->layout.get_period() * max;
+ if (len > tr->length)
+ len = tr->length;
+
+ uint64_t offset = tr->offset + tr->length - len;
+ Striper::file_to_extents(cct, tr->ino, &tr->layout, offset, len, 0, extents);
+ tr->uncommitted += extents.size();
+ tr->length -= len;
+ }
+
+ trl.unlock();
+
+ // Issue objecter ops outside tr->lock to avoid lock dependency loop
+ for (const auto& p : extents) {
+ vector<OSDOp> ops(1);
+ ops[0].op.op = CEPH_OSD_OP_TRIMTRUNC;
+ ops[0].op.extent.truncate_size = p.offset;
+ ops[0].op.extent.truncate_seq = tr->truncate_seq;
+ objecter->_modify(p.oid, p.oloc, ops, tr->mtime, tr->snapc, tr->flags,
+ new C_OnFinisher(new C_TruncRange(this, tr), finisher));
+ }
+}
Code Review / stor4nfv.git / blobdiff