Simplified distributed block storage with strong consistency, like in Ceph
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 
 
 

278 lines
9.7 KiB

  1. // Copyright (c) Vitaliy Filippov, 2019+
  2. // License: VNPL-1.1 (see README.md for details)
  3. #include "blockstore_impl.h"
  4. // Stabilize small write:
  5. // 1) Copy data from the journal to the data device
  6. // 2) Increase version on the metadata device and sync it
  7. // 3) Advance clean_db entry's version, clear previous journal entries
  8. //
  9. // This makes 1 4K small write+sync look like:
  10. // 512b+4K (journal) + sync + 512b (journal) + sync + 4K (data) [+ sync?] + 512b (metadata) + sync.
  11. // WA = 2.375. It's not the best, SSD FTL-like redirect-write could probably be lower
  12. // even with defragmentation. But it's fixed and it's still better than in Ceph. :)
  13. // except for HDD-only clusters, because each write results in 3 seeks.
  14. // Stabilize big write:
  15. // 1) Copy metadata from the journal to the metadata device
  16. // 2) Move dirty_db entry to clean_db and clear previous journal entries
  17. //
  18. // This makes 1 128K big write+sync look like:
  19. // 128K (data) + sync + 512b (journal) + sync + 512b (journal) + sync + 512b (metadata) + sync.
  20. // WA = 1.012. Very good :)
  21. // Stabilize delete:
  22. // 1) Remove metadata entry and sync it
  23. // 2) Remove dirty_db entry and clear previous journal entries
  24. // We have 2 problems here:
  25. // - In the cluster environment, we must store the "tombstones" of deleted objects until
  26. // all replicas (not just quorum) agrees about their deletion. That is, "stabilize" is
  27. // not possible for deletes in degraded placement groups
  28. // - With simple "fixed" metadata tables we can't just clear the metadata entry of the latest
  29. // object version. We must clear all previous entries, too.
  30. // FIXME Fix both problems - probably, by switching from "fixed" metadata tables to "dynamic"
  31. // AND We must do it in batches, for the sake of reduced fsync call count
  32. // AND We must know what we stabilize. Basic workflow is like:
  33. // 1) primary OSD receives sync request
  34. // 2) it submits syncs to blockstore and peers
  35. // 3) after everyone acks sync it acks sync to the client
  36. // 4) after a while it takes his synced object list and sends stabilize requests
  37. // to peers and to its own blockstore, thus freeing the old version
  38. int blockstore_impl_t::dequeue_stable(blockstore_op_t *op)
  39. {
  40. if (PRIV(op)->op_state)
  41. {
  42. return continue_stable(op);
  43. }
  44. obj_ver_id* v;
  45. int i, todo = 0;
  46. for (i = 0, v = (obj_ver_id*)op->buf; i < op->len; i++, v++)
  47. {
  48. auto dirty_it = dirty_db.find(*v);
  49. if (dirty_it == dirty_db.end())
  50. {
  51. auto clean_it = clean_db.find(v->oid);
  52. if (clean_it == clean_db.end() || clean_it->second.version < v->version)
  53. {
  54. // No such object version
  55. op->retval = -ENOENT;
  56. FINISH_OP(op);
  57. return 2;
  58. }
  59. else
  60. {
  61. // Already stable
  62. }
  63. }
  64. else if (IS_IN_FLIGHT(dirty_it->second.state))
  65. {
  66. // Object write is still in progress. Wait until the write request completes
  67. return 0;
  68. }
  69. else if (!IS_SYNCED(dirty_it->second.state))
  70. {
  71. // Object not synced yet. Caller must sync it first
  72. op->retval = -EBUSY;
  73. FINISH_OP(op);
  74. return 2;
  75. }
  76. else if (!IS_STABLE(dirty_it->second.state))
  77. {
  78. todo++;
  79. }
  80. }
  81. if (!todo)
  82. {
  83. // Already stable
  84. op->retval = 0;
  85. FINISH_OP(op);
  86. return 2;
  87. }
  88. // Check journal space
  89. blockstore_journal_check_t space_check(this);
  90. if (!space_check.check_available(op, todo, sizeof(journal_entry_stable), 0))
  91. {
  92. return 0;
  93. }
  94. // There is sufficient space. Get SQEs
  95. struct io_uring_sqe *sqe[space_check.sectors_to_write];
  96. for (i = 0; i < space_check.sectors_to_write; i++)
  97. {
  98. BS_SUBMIT_GET_SQE_DECL(sqe[i]);
  99. }
  100. // Prepare and submit journal entries
  101. auto cb = [this, op](ring_data_t *data) { handle_stable_event(data, op); };
  102. int s = 0, cur_sector = -1;
  103. for (i = 0, v = (obj_ver_id*)op->buf; i < op->len; i++, v++)
  104. {
  105. // FIXME: Only stabilize versions that aren't stable yet
  106. if (!journal.entry_fits(sizeof(journal_entry_stable)) &&
  107. journal.sector_info[journal.cur_sector].dirty)
  108. {
  109. if (cur_sector == -1)
  110. PRIV(op)->min_flushed_journal_sector = 1 + journal.cur_sector;
  111. prepare_journal_sector_write(journal, journal.cur_sector, sqe[s++], cb);
  112. cur_sector = journal.cur_sector;
  113. }
  114. journal_entry_stable *je = (journal_entry_stable*)
  115. prefill_single_journal_entry(journal, JE_STABLE, sizeof(journal_entry_stable));
  116. je->oid = v->oid;
  117. je->version = v->version;
  118. je->crc32 = je_crc32((journal_entry*)je);
  119. journal.crc32_last = je->crc32;
  120. }
  121. prepare_journal_sector_write(journal, journal.cur_sector, sqe[s++], cb);
  122. assert(s == space_check.sectors_to_write);
  123. if (cur_sector == -1)
  124. PRIV(op)->min_flushed_journal_sector = 1 + journal.cur_sector;
  125. PRIV(op)->max_flushed_journal_sector = 1 + journal.cur_sector;
  126. PRIV(op)->pending_ops = s;
  127. PRIV(op)->op_state = 1;
  128. return 1;
  129. }
  130. int blockstore_impl_t::continue_stable(blockstore_op_t *op)
  131. {
  132. if (PRIV(op)->op_state == 2)
  133. goto resume_2;
  134. else if (PRIV(op)->op_state == 3)
  135. goto resume_3;
  136. else if (PRIV(op)->op_state == 5)
  137. goto resume_5;
  138. else
  139. return 1;
  140. resume_2:
  141. // Release used journal sectors
  142. release_journal_sectors(op);
  143. resume_3:
  144. if (!disable_journal_fsync)
  145. {
  146. io_uring_sqe *sqe;
  147. BS_SUBMIT_GET_SQE_DECL(sqe);
  148. ring_data_t *data = ((ring_data_t*)sqe->user_data);
  149. my_uring_prep_fsync(sqe, journal.fd, IORING_FSYNC_DATASYNC);
  150. data->iov = { 0 };
  151. data->callback = [this, op](ring_data_t *data) { handle_stable_event(data, op); };
  152. PRIV(op)->min_flushed_journal_sector = PRIV(op)->max_flushed_journal_sector = 0;
  153. PRIV(op)->pending_ops = 1;
  154. PRIV(op)->op_state = 4;
  155. return 1;
  156. }
  157. resume_5:
  158. // Mark dirty_db entries as stable, acknowledge op completion
  159. obj_ver_id* v;
  160. int i;
  161. for (i = 0, v = (obj_ver_id*)op->buf; i < op->len; i++, v++)
  162. {
  163. // Mark all dirty_db entries up to op->version as stable
  164. #ifdef BLOCKSTORE_DEBUG
  165. printf("Stabilize %lx:%lx v%lu\n", v->oid.inode, v->oid.stripe, v->version);
  166. #endif
  167. mark_stable(*v);
  168. }
  169. // Acknowledge op
  170. op->retval = 0;
  171. FINISH_OP(op);
  172. return 2;
  173. }
  174. void blockstore_impl_t::mark_stable(const obj_ver_id & v, bool forget_dirty)
  175. {
  176. auto dirty_it = dirty_db.find(v);
  177. if (dirty_it != dirty_db.end())
  178. {
  179. while (1)
  180. {
  181. bool was_stable = IS_STABLE(dirty_it->second.state);
  182. if ((dirty_it->second.state & BS_ST_WORKFLOW_MASK) == BS_ST_SYNCED)
  183. {
  184. dirty_it->second.state = (dirty_it->second.state & ~BS_ST_WORKFLOW_MASK) | BS_ST_STABLE;
  185. // Allocations and deletions are counted when they're stabilized
  186. if (IS_BIG_WRITE(dirty_it->second.state))
  187. {
  188. int exists = -1;
  189. if (dirty_it != dirty_db.begin())
  190. {
  191. auto prev_it = dirty_it;
  192. prev_it--;
  193. if (prev_it->first.oid == v.oid)
  194. {
  195. exists = IS_DELETE(prev_it->second.state) ? 0 : 1;
  196. }
  197. }
  198. if (exists == -1)
  199. {
  200. auto clean_it = clean_db.find(v.oid);
  201. exists = clean_it != clean_db.end() ? 1 : 0;
  202. }
  203. if (!exists)
  204. {
  205. inode_space_stats[dirty_it->first.oid.inode] += block_size;
  206. }
  207. }
  208. else if (IS_DELETE(dirty_it->second.state))
  209. {
  210. inode_space_stats[dirty_it->first.oid.inode] -= block_size;
  211. }
  212. }
  213. if (forget_dirty && (IS_BIG_WRITE(dirty_it->second.state) ||
  214. IS_DELETE(dirty_it->second.state)))
  215. {
  216. // Big write overrides all previous dirty entries
  217. auto erase_end = dirty_it;
  218. while (dirty_it != dirty_db.begin())
  219. {
  220. dirty_it--;
  221. if (dirty_it->first.oid != v.oid)
  222. {
  223. dirty_it++;
  224. break;
  225. }
  226. }
  227. auto clean_it = clean_db.find(v.oid);
  228. uint64_t clean_loc = clean_it != clean_db.end()
  229. ? clean_it->second.location : UINT64_MAX;
  230. erase_dirty(dirty_it, erase_end, clean_loc);
  231. break;
  232. }
  233. if (was_stable || dirty_it == dirty_db.begin())
  234. {
  235. break;
  236. }
  237. dirty_it--;
  238. if (dirty_it->first.oid != v.oid)
  239. {
  240. break;
  241. }
  242. }
  243. flusher->enqueue_flush(v);
  244. }
  245. auto unstab_it = unstable_writes.find(v.oid);
  246. if (unstab_it != unstable_writes.end() &&
  247. unstab_it->second <= v.version)
  248. {
  249. unstable_writes.erase(unstab_it);
  250. }
  251. }
  252. void blockstore_impl_t::handle_stable_event(ring_data_t *data, blockstore_op_t *op)
  253. {
  254. live = true;
  255. if (data->res != data->iov.iov_len)
  256. {
  257. throw std::runtime_error(
  258. "write operation failed ("+std::to_string(data->res)+" != "+std::to_string(data->iov.iov_len)+
  259. "). in-memory state is corrupted. AAAAAAAaaaaaaaaa!!!111"
  260. );
  261. }
  262. PRIV(op)->pending_ops--;
  263. if (PRIV(op)->pending_ops == 0)
  264. {
  265. PRIV(op)->op_state++;
  266. ringloop->wakeup();
  267. }
  268. }