Line data Source code
1 : use std::{ffi::CString, sync::Arc};
2 :
3 : use byteorder::{LittleEndian, WriteBytesExt};
4 : use crc32c::crc32c_append;
5 : use parking_lot::{Mutex, MutexGuard};
6 : use postgres_ffi::{
7 : pg_constants::{
8 : RM_LOGICALMSG_ID, XLOG_LOGICAL_MESSAGE, XLP_LONG_HEADER, XLR_BLOCK_ID_DATA_LONG,
9 : XLR_BLOCK_ID_DATA_SHORT,
10 : },
11 : v16::{
12 : wal_craft_test_export::{XLogLongPageHeaderData, XLogPageHeaderData, XLOG_PAGE_MAGIC},
13 : xlog_utils::{
14 : XLogSegNoOffsetToRecPtr, XlLogicalMessage, XLOG_RECORD_CRC_OFFS,
15 : XLOG_SIZE_OF_XLOG_LONG_PHD, XLOG_SIZE_OF_XLOG_RECORD, XLOG_SIZE_OF_XLOG_SHORT_PHD,
16 : XLP_FIRST_IS_CONTRECORD,
17 : },
18 : XLogRecord,
19 : },
20 : WAL_SEGMENT_SIZE, XLOG_BLCKSZ,
21 : };
22 : use utils::lsn::Lsn;
23 :
24 : use super::block_storage::BlockStorage;
25 :
26 : /// Simulation implementation of walproposer WAL storage.
27 : pub struct DiskWalProposer {
28 : state: Mutex<State>,
29 : }
30 :
31 : impl DiskWalProposer {
32 74015 : pub fn new() -> Arc<DiskWalProposer> {
33 74015 : Arc::new(DiskWalProposer {
34 74015 : state: Mutex::new(State {
35 74015 : internal_available_lsn: Lsn(0),
36 74015 : prev_lsn: Lsn(0),
37 74015 : disk: BlockStorage::new(),
38 74015 : }),
39 74015 : })
40 74015 : }
41 :
42 81012 : pub fn lock(&self) -> MutexGuard<State> {
43 81012 : self.state.lock()
44 81012 : }
45 : }
46 :
47 : pub struct State {
48 : // flush_lsn
49 : internal_available_lsn: Lsn,
50 : // needed for WAL generation
51 : prev_lsn: Lsn,
52 : // actual WAL storage
53 : disk: BlockStorage,
54 : }
55 :
56 : impl State {
57 29244 : pub fn read(&self, pos: u64, buf: &mut [u8]) {
58 29244 : self.disk.read(pos, buf);
59 29244 : // TODO: fail on reading uninitialized data
60 29244 : }
61 :
62 141155 : pub fn write(&mut self, pos: u64, buf: &[u8]) {
63 141155 : self.disk.write(pos, buf);
64 141155 : }
65 :
66 : /// Update the internal available LSN to the given value.
67 3022 : pub fn reset_to(&mut self, lsn: Lsn) {
68 3022 : self.internal_available_lsn = lsn;
69 3022 : }
70 :
71 : /// Get current LSN.
72 12257 : pub fn flush_rec_ptr(&self) -> Lsn {
73 12257 : self.internal_available_lsn
74 12257 : }
75 :
76 : /// Generate a new WAL record at the current LSN.
77 34766 : pub fn insert_logical_message(&mut self, prefix: &str, msg: &[u8]) -> anyhow::Result<()> {
78 34766 : let prefix_cstr = CString::new(prefix)?;
79 34766 : let prefix_bytes = prefix_cstr.as_bytes_with_nul();
80 34766 :
81 34766 : let lm = XlLogicalMessage {
82 34766 : db_id: 0,
83 34766 : transactional: 0,
84 34766 : prefix_size: prefix_bytes.len() as ::std::os::raw::c_ulong,
85 34766 : message_size: msg.len() as ::std::os::raw::c_ulong,
86 34766 : };
87 34766 :
88 34766 : let record_bytes = lm.encode();
89 34766 : let rdatas: Vec<&[u8]> = vec![&record_bytes, prefix_bytes, msg];
90 34766 : insert_wal_record(self, rdatas, RM_LOGICALMSG_ID, XLOG_LOGICAL_MESSAGE)
91 34766 : }
92 : }
93 :
94 34766 : fn insert_wal_record(
95 34766 : state: &mut State,
96 34766 : rdatas: Vec<&[u8]>,
97 34766 : rmid: u8,
98 34766 : info: u8,
99 34766 : ) -> anyhow::Result<()> {
100 34766 : // bytes right after the header, in the same rdata block
101 34766 : let mut scratch = Vec::new();
102 104298 : let mainrdata_len: usize = rdatas.iter().map(|rdata| rdata.len()).sum();
103 34766 :
104 34766 : if mainrdata_len > 0 {
105 34766 : if mainrdata_len > 255 {
106 0 : scratch.push(XLR_BLOCK_ID_DATA_LONG);
107 0 : // TODO: verify endiness
108 0 : let _ = scratch.write_u32::<LittleEndian>(mainrdata_len as u32);
109 34766 : } else {
110 34766 : scratch.push(XLR_BLOCK_ID_DATA_SHORT);
111 34766 : scratch.push(mainrdata_len as u8);
112 34766 : }
113 0 : }
114 :
115 34766 : let total_len: u32 = (XLOG_SIZE_OF_XLOG_RECORD + scratch.len() + mainrdata_len) as u32;
116 34766 : let size = maxalign(total_len);
117 34766 : assert!(size as usize > XLOG_SIZE_OF_XLOG_RECORD);
118 :
119 34766 : let start_bytepos = recptr_to_bytepos(state.internal_available_lsn);
120 34766 : let end_bytepos = start_bytepos + size as u64;
121 34766 :
122 34766 : let start_recptr = bytepos_to_recptr(start_bytepos);
123 34766 : let end_recptr = bytepos_to_recptr(end_bytepos);
124 34766 :
125 34766 : assert!(recptr_to_bytepos(start_recptr) == start_bytepos);
126 34766 : assert!(recptr_to_bytepos(end_recptr) == end_bytepos);
127 :
128 34766 : let mut crc = crc32c_append(0, &scratch);
129 139064 : for rdata in &rdatas {
130 104298 : crc = crc32c_append(crc, rdata);
131 104298 : }
132 :
133 34766 : let mut header = XLogRecord {
134 34766 : xl_tot_len: total_len,
135 34766 : xl_xid: 0,
136 34766 : xl_prev: state.prev_lsn.0,
137 34766 : xl_info: info,
138 34766 : xl_rmid: rmid,
139 34766 : __bindgen_padding_0: [0u8; 2usize],
140 34766 : xl_crc: crc,
141 34766 : };
142 :
143 : // now we have the header and can finish the crc
144 34766 : let header_bytes = header.encode()?;
145 34766 : let crc = crc32c_append(crc, &header_bytes[0..XLOG_RECORD_CRC_OFFS]);
146 34766 : header.xl_crc = crc;
147 :
148 34766 : let mut header_bytes = header.encode()?.to_vec();
149 34766 : assert!(header_bytes.len() == XLOG_SIZE_OF_XLOG_RECORD);
150 :
151 34766 : header_bytes.extend_from_slice(&scratch);
152 34766 :
153 34766 : // finish rdatas
154 34766 : let mut rdatas = rdatas;
155 34766 : rdatas.insert(0, &header_bytes);
156 34766 :
157 34766 : write_walrecord_to_disk(state, total_len as u64, rdatas, start_recptr, end_recptr)?;
158 :
159 34766 : state.internal_available_lsn = end_recptr;
160 34766 : state.prev_lsn = start_recptr;
161 34766 : Ok(())
162 34766 : }
163 :
164 34766 : fn write_walrecord_to_disk(
165 34766 : state: &mut State,
166 34766 : total_len: u64,
167 34766 : rdatas: Vec<&[u8]>,
168 34766 : start: Lsn,
169 34766 : end: Lsn,
170 34766 : ) -> anyhow::Result<()> {
171 34766 : let mut curr_ptr = start;
172 34766 : let mut freespace = insert_freespace(curr_ptr);
173 34766 : let mut written: usize = 0;
174 34766 :
175 34766 : assert!(freespace >= std::mem::size_of::<u32>());
176 :
177 173830 : for mut rdata in rdatas {
178 139248 : while rdata.len() >= freespace {
179 184 : assert!(
180 184 : curr_ptr.segment_offset(WAL_SEGMENT_SIZE) >= XLOG_SIZE_OF_XLOG_SHORT_PHD
181 0 : || freespace == 0
182 : );
183 :
184 184 : state.write(curr_ptr.0, &rdata[..freespace]);
185 184 : rdata = &rdata[freespace..];
186 184 : written += freespace;
187 184 : curr_ptr = Lsn(curr_ptr.0 + freespace as u64);
188 184 :
189 184 : let mut new_page = XLogPageHeaderData {
190 184 : xlp_magic: XLOG_PAGE_MAGIC as u16,
191 184 : xlp_info: XLP_BKP_REMOVABLE,
192 184 : xlp_tli: 1,
193 184 : xlp_pageaddr: curr_ptr.0,
194 184 : xlp_rem_len: (total_len - written as u64) as u32,
195 184 : ..Default::default() // Put 0 in padding fields.
196 184 : };
197 184 : if new_page.xlp_rem_len > 0 {
198 166 : new_page.xlp_info |= XLP_FIRST_IS_CONTRECORD;
199 166 : }
200 :
201 184 : if curr_ptr.segment_offset(WAL_SEGMENT_SIZE) == 0 {
202 0 : new_page.xlp_info |= XLP_LONG_HEADER;
203 0 : let long_page = XLogLongPageHeaderData {
204 0 : std: new_page,
205 0 : xlp_sysid: 0,
206 0 : xlp_seg_size: WAL_SEGMENT_SIZE as u32,
207 0 : xlp_xlog_blcksz: XLOG_BLCKSZ as u32,
208 0 : };
209 0 : let header_bytes = long_page.encode()?;
210 0 : assert!(header_bytes.len() == XLOG_SIZE_OF_XLOG_LONG_PHD);
211 0 : state.write(curr_ptr.0, &header_bytes);
212 0 : curr_ptr = Lsn(curr_ptr.0 + header_bytes.len() as u64);
213 : } else {
214 184 : let header_bytes = new_page.encode()?;
215 184 : assert!(header_bytes.len() == XLOG_SIZE_OF_XLOG_SHORT_PHD);
216 184 : state.write(curr_ptr.0, &header_bytes);
217 184 : curr_ptr = Lsn(curr_ptr.0 + header_bytes.len() as u64);
218 : }
219 184 : freespace = insert_freespace(curr_ptr);
220 : }
221 :
222 139064 : assert!(
223 139064 : curr_ptr.segment_offset(WAL_SEGMENT_SIZE) >= XLOG_SIZE_OF_XLOG_SHORT_PHD
224 0 : || rdata.is_empty()
225 : );
226 139064 : state.write(curr_ptr.0, rdata);
227 139064 : curr_ptr = Lsn(curr_ptr.0 + rdata.len() as u64);
228 139064 : written += rdata.len();
229 139064 : freespace -= rdata.len();
230 : }
231 :
232 34766 : assert!(written == total_len as usize);
233 34766 : curr_ptr.0 = maxalign(curr_ptr.0);
234 34766 : assert!(curr_ptr == end);
235 34766 : Ok(())
236 34766 : }
237 :
238 69532 : fn maxalign<T>(size: T) -> T
239 69532 : where
240 69532 : T: std::ops::BitAnd<Output = T>
241 69532 : + std::ops::Add<Output = T>
242 69532 : + std::ops::Not<Output = T>
243 69532 : + From<u8>,
244 69532 : {
245 69532 : (size + T::from(7)) & !T::from(7)
246 69532 : }
247 :
248 34950 : fn insert_freespace(ptr: Lsn) -> usize {
249 34950 : if ptr.block_offset() == 0 {
250 0 : 0
251 : } else {
252 34950 : (XLOG_BLCKSZ as u64 - ptr.block_offset()) as usize
253 : }
254 34950 : }
255 :
256 : const XLP_BKP_REMOVABLE: u16 = 0x0004;
257 : const USABLE_BYTES_IN_PAGE: u64 = (XLOG_BLCKSZ - XLOG_SIZE_OF_XLOG_SHORT_PHD) as u64;
258 : const USABLE_BYTES_IN_SEGMENT: u64 = ((WAL_SEGMENT_SIZE / XLOG_BLCKSZ) as u64
259 : * USABLE_BYTES_IN_PAGE)
260 : - (XLOG_SIZE_OF_XLOG_RECORD - XLOG_SIZE_OF_XLOG_SHORT_PHD) as u64;
261 :
262 69532 : fn bytepos_to_recptr(bytepos: u64) -> Lsn {
263 69532 : let fullsegs = bytepos / USABLE_BYTES_IN_SEGMENT;
264 69532 : let mut bytesleft = bytepos % USABLE_BYTES_IN_SEGMENT;
265 :
266 69532 : let seg_offset = if bytesleft < (XLOG_BLCKSZ - XLOG_SIZE_OF_XLOG_SHORT_PHD) as u64 {
267 : // fits on first page of segment
268 0 : bytesleft + XLOG_SIZE_OF_XLOG_SHORT_PHD as u64
269 : } else {
270 : // account for the first page on segment with long header
271 69532 : bytesleft -= (XLOG_BLCKSZ - XLOG_SIZE_OF_XLOG_SHORT_PHD) as u64;
272 69532 : let fullpages = bytesleft / USABLE_BYTES_IN_PAGE;
273 69532 : bytesleft %= USABLE_BYTES_IN_PAGE;
274 69532 :
275 69532 : XLOG_BLCKSZ as u64
276 69532 : + fullpages * XLOG_BLCKSZ as u64
277 69532 : + bytesleft
278 69532 : + XLOG_SIZE_OF_XLOG_SHORT_PHD as u64
279 : };
280 :
281 69532 : Lsn(XLogSegNoOffsetToRecPtr(
282 69532 : fullsegs,
283 69532 : seg_offset as u32,
284 69532 : WAL_SEGMENT_SIZE,
285 69532 : ))
286 69532 : }
287 :
288 104298 : fn recptr_to_bytepos(ptr: Lsn) -> u64 {
289 104298 : let fullsegs = ptr.segment_number(WAL_SEGMENT_SIZE);
290 104298 : let offset = ptr.segment_offset(WAL_SEGMENT_SIZE) as u64;
291 104298 :
292 104298 : let fullpages = offset / XLOG_BLCKSZ as u64;
293 104298 : let offset = offset % XLOG_BLCKSZ as u64;
294 104298 :
295 104298 : if fullpages == 0 {
296 0 : fullsegs * USABLE_BYTES_IN_SEGMENT
297 0 : + if offset > 0 {
298 0 : assert!(offset >= XLOG_SIZE_OF_XLOG_SHORT_PHD as u64);
299 0 : offset - XLOG_SIZE_OF_XLOG_SHORT_PHD as u64
300 : } else {
301 0 : 0
302 : }
303 : } else {
304 104298 : fullsegs * USABLE_BYTES_IN_SEGMENT
305 104298 : + (XLOG_BLCKSZ - XLOG_SIZE_OF_XLOG_SHORT_PHD) as u64
306 104298 : + (fullpages - 1) * USABLE_BYTES_IN_PAGE
307 104298 : + if offset > 0 {
308 104298 : assert!(offset >= XLOG_SIZE_OF_XLOG_SHORT_PHD as u64);
309 104298 : offset - XLOG_SIZE_OF_XLOG_SHORT_PHD as u64
310 : } else {
311 0 : 0
312 : }
313 : }
314 104298 : }
|
