walWrite.c 11.0 KB
Newer Older
陶建辉(Jeff)'s avatar
陶建辉(Jeff) 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16
/*
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
 *
 * This program is free software: you can use, redistribute, and/or modify
 * it under the terms of the GNU Affero General Public License, version 3
 * or later ("AGPL"), as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */

#include "os.h"
S
Shengliang Guan 已提交
17
#include "taoserror.h"
18
#include "tchecksum.h"
S
Shengliang Guan 已提交
19
#include "walInt.h"
陶建辉(Jeff)'s avatar
陶建辉(Jeff) 已提交
20

L
Liu Jicong 已提交
21
int32_t walCommit(SWal *pWal, int64_t ver) {
L
Liu Jicong 已提交
22 23
  ASSERT(pWal->vers.commitVer >= pWal->vers.snapshotVer);
  ASSERT(pWal->vers.commitVer <= pWal->vers.lastVer);
L
Liu Jicong 已提交
24
  if (ver < pWal->vers.commitVer || ver > pWal->vers.lastVer) {
L
Liu Jicong 已提交
25
    terrno = TSDB_CODE_WAL_INVALID_VER;
L
Liu Jicong 已提交
26 27
    return -1;
  }
L
Liu Jicong 已提交
28
  pWal->vers.commitVer = ver;
L
Liu Jicong 已提交
29 30 31 32
  return 0;
}

int32_t walRollback(SWal *pWal, int64_t ver) {
L
Liu Jicong 已提交
33
  int  code;
L
Liu Jicong 已提交
34
  char fnameStr[WAL_FILE_LEN];
L
Liu Jicong 已提交
35
  if (ver > pWal->vers.lastVer || ver < pWal->vers.commitVer) {
L
Liu Jicong 已提交
36
    terrno = TSDB_CODE_WAL_INVALID_VER;
L
Liu Jicong 已提交
37 38
    return -1;
  }
wafwerar's avatar
wafwerar 已提交
39
  taosThreadMutexLock(&pWal->mutex);
L
Liu Jicong 已提交
40

L
Liu Jicong 已提交
41 42
  // find correct file
  if (ver < walGetLastFileFirstVer(pWal)) {
L
Liu Jicong 已提交
43 44 45
    // change current files
    code = walChangeWrite(pWal, ver);
    if (code < 0) {
L
Liu Jicong 已提交
46 47 48
      return -1;
    }

L
Liu Jicong 已提交
49
    // delete files
L
Liu Jicong 已提交
50
    int fileSetSize = taosArrayGetSize(pWal->fileInfoSet);
L
Liu Jicong 已提交
51
    for (int i = pWal->writeCur; i < fileSetSize; i++) {
L
Liu Jicong 已提交
52
      walBuildLogName(pWal, ((SWalFileInfo *)taosArrayGet(pWal->fileInfoSet, i))->firstVer, fnameStr);
53
      taosRemoveFile(fnameStr);
L
Liu Jicong 已提交
54
      walBuildIdxName(pWal, ((SWalFileInfo *)taosArrayGet(pWal->fileInfoSet, i))->firstVer, fnameStr);
55
      taosRemoveFile(fnameStr);
L
Liu Jicong 已提交
56
    }
L
Liu Jicong 已提交
57
    // pop from fileInfoSet
L
Liu Jicong 已提交
58 59 60 61
    taosArraySetSize(pWal->fileInfoSet, pWal->writeCur + 1);
  }

  walBuildIdxName(pWal, walGetCurFileFirstVer(pWal), fnameStr);
62
  TdFilePtr pIdxTFile = taosOpenFile(fnameStr, TD_FILE_WRITE | TD_FILE_READ);
L
Liu Jicong 已提交
63

L
Liu Jicong 已提交
64
  // TODO:change to deserialize function
65
  if (pIdxTFile == NULL) {
wafwerar's avatar
wafwerar 已提交
66
    taosThreadMutexUnlock(&pWal->mutex);
L
Liu Jicong 已提交
67 68
    return -1;
  }
L
Liu Jicong 已提交
69
  int64_t idxOff = walGetVerIdxOffset(pWal, ver);
70
  code = taosLSeekFile(pIdxTFile, idxOff, SEEK_SET);
L
Liu Jicong 已提交
71
  if (code < 0) {
wafwerar's avatar
wafwerar 已提交
72
    taosThreadMutexUnlock(&pWal->mutex);
L
Liu Jicong 已提交
73 74
    return -1;
  }
L
Liu Jicong 已提交
75 76
  // read idx file and get log file pos
  // TODO:change to deserialize function
L
Liu Jicong 已提交
77
  SWalIdxEntry entry;
78
  if (taosReadFile(pIdxTFile, &entry, sizeof(SWalIdxEntry)) != sizeof(SWalIdxEntry)) {
wafwerar's avatar
wafwerar 已提交
79
    taosThreadMutexUnlock(&pWal->mutex);
L
Liu Jicong 已提交
80 81 82 83 84
    return -1;
  }
  ASSERT(entry.ver == ver);

  walBuildLogName(pWal, walGetCurFileFirstVer(pWal), fnameStr);
85
  TdFilePtr pLogTFile = taosOpenFile(fnameStr, TD_FILE_WRITE | TD_FILE_READ);
86
  if (pLogTFile == NULL) {
L
Liu Jicong 已提交
87
    // TODO
wafwerar's avatar
wafwerar 已提交
88
    taosThreadMutexUnlock(&pWal->mutex);
L
Liu Jicong 已提交
89 90
    return -1;
  }
91
  code = taosLSeekFile(pLogTFile, entry.offset, SEEK_SET);
L
Liu Jicong 已提交
92 93
  if (code < 0) {
    // TODO
wafwerar's avatar
wafwerar 已提交
94
    taosThreadMutexUnlock(&pWal->mutex);
L
Liu Jicong 已提交
95 96
    return -1;
  }
L
Liu Jicong 已提交
97
  // validate offset
L
Liu Jicong 已提交
98
  SWalHead head;
99 100
  ASSERT(taosValidFile(pLogTFile));
  int size = taosReadFile(pLogTFile, &head, sizeof(SWalHead));
L
Liu Jicong 已提交
101
  if (size != sizeof(SWalHead)) {
L
Liu Jicong 已提交
102 103 104 105 106
    return -1;
  }
  code = walValidHeadCksum(&head);

  ASSERT(code == 0);
L
Liu Jicong 已提交
107
  if (code != 0) {
L
Liu Jicong 已提交
108 109
    return -1;
  }
L
Liu Jicong 已提交
110 111
  if (head.head.version != ver) {
    // TODO
L
Liu Jicong 已提交
112 113
    return -1;
  }
L
Liu Jicong 已提交
114
  // truncate old files
115
  code = taosFtruncateFile(pLogTFile, entry.offset);
L
Liu Jicong 已提交
116
  if (code < 0) {
L
Liu Jicong 已提交
117 118
    return -1;
  }
119
  code = taosFtruncateFile(pIdxTFile, idxOff);
L
Liu Jicong 已提交
120
  if (code < 0) {
L
Liu Jicong 已提交
121 122
    return -1;
  }
L
Liu Jicong 已提交
123
  pWal->vers.lastVer = ver - 1;
L
Liu Jicong 已提交
124 125
  ((SWalFileInfo *)taosArrayGetLast(pWal->fileInfoSet))->lastVer = ver - 1;
  ((SWalFileInfo *)taosArrayGetLast(pWal->fileInfoSet))->fileSize = entry.offset;
L
Liu Jicong 已提交
126

L
Liu Jicong 已提交
127
  // unlock
wafwerar's avatar
wafwerar 已提交
128
  taosThreadMutexUnlock(&pWal->mutex);
L
Liu Jicong 已提交
129 130
  return 0;
}
L
Liu Jicong 已提交
131

L
Liu Jicong 已提交
132
int32_t walBeginSnapshot(SWal *pWal, int64_t ver) {
L
Liu Jicong 已提交
133
  pWal->vers.verInSnapshotting = ver;
L
Liu Jicong 已提交
134 135
  // check file rolling
  if (pWal->cfg.retentionPeriod == 0) {
L
Liu Jicong 已提交
136 137 138 139 140 141
    walRoll(pWal);
  }

  return 0;
}

L
Liu Jicong 已提交
142
int32_t walEndSnapshot(SWal *pWal) {
L
Liu Jicong 已提交
143
  int64_t ver = pWal->vers.verInSnapshotting;
S
Shengliang Guan 已提交
144
  if (ver == -1) return 0;
L
Liu Jicong 已提交
145

L
Liu Jicong 已提交
146
  pWal->vers.snapshotVer = ver;
L
Liu Jicong 已提交
147 148
  int ts = taosGetTimestampSec();

L
Liu Jicong 已提交
149 150 151
  int          deleteCnt = 0;
  int64_t      newTotSize = pWal->totSize;
  SWalFileInfo tmp;
L
Liu Jicong 已提交
152
  tmp.firstVer = ver;
L
Liu Jicong 已提交
153
  // find files safe to delete
L
Liu Jicong 已提交
154
  SWalFileInfo *pInfo = taosArraySearch(pWal->fileInfoSet, &tmp, compareWalFileInfo, TD_LE);
L
Liu Jicong 已提交
155
  if (ver >= pInfo->lastVer) {
L
Liu Jicong 已提交
156 157
    pInfo++;
  }
L
Liu Jicong 已提交
158
  // iterate files, until the searched result
L
Liu Jicong 已提交
159
  for (SWalFileInfo *iter = pWal->fileInfoSet->pData; iter < pInfo; iter++) {
L
fix  
Liu Jicong 已提交
160 161
    if ((pWal->cfg.retentionSize != -1 && pWal->totSize > pWal->cfg.retentionSize) ||
        (pWal->cfg.retentionPeriod != -1 && iter->closeTs + pWal->cfg.retentionPeriod > ts)) {
L
Liu Jicong 已提交
162
      // delete according to file size or close time
L
Liu Jicong 已提交
163 164 165 166 167
      deleteCnt++;
      newTotSize -= iter->fileSize;
    }
  }
  char fnameStr[WAL_FILE_LEN];
L
Liu Jicong 已提交
168 169
  // remove file
  for (int i = 0; i < deleteCnt; i++) {
L
Liu Jicong 已提交
170
    SWalFileInfo *pInfo = taosArrayGet(pWal->fileInfoSet, i);
L
Liu Jicong 已提交
171
    walBuildLogName(pWal, pInfo->firstVer, fnameStr);
172
    taosRemoveFile(fnameStr);
L
Liu Jicong 已提交
173
    walBuildIdxName(pWal, pInfo->firstVer, fnameStr);
174
    taosRemoveFile(fnameStr);
L
Liu Jicong 已提交
175 176
  }

L
Liu Jicong 已提交
177 178 179
  // make new array, remove files
  taosArrayPopFrontBatch(pWal->fileInfoSet, deleteCnt);
  if (taosArrayGetSize(pWal->fileInfoSet) == 0) {
L
Liu Jicong 已提交
180
    pWal->writeCur = -1;
L
Liu Jicong 已提交
181
    pWal->vers.firstVer = -1;
L
Liu Jicong 已提交
182
  } else {
L
Liu Jicong 已提交
183
    pWal->vers.firstVer = ((SWalFileInfo *)taosArrayGet(pWal->fileInfoSet, 0))->firstVer;
L
Liu Jicong 已提交
184
  }
L
Liu Jicong 已提交
185
  pWal->writeCur = taosArrayGetSize(pWal->fileInfoSet) - 1;
L
Liu Jicong 已提交
186
  pWal->totSize = newTotSize;
L
Liu Jicong 已提交
187
  pWal->vers.verInSnapshotting = -1;
L
Liu Jicong 已提交
188

L
Liu Jicong 已提交
189
  // save snapshot ver, commit ver
L
Liu Jicong 已提交
190
  int code = walSaveMeta(pWal);
L
Liu Jicong 已提交
191
  if (code < 0) {
L
Liu Jicong 已提交
192 193 194 195 196 197
    return -1;
  }

  return 0;
}

L
Liu Jicong 已提交
198
int walRoll(SWal *pWal) {
L
Liu Jicong 已提交
199
  int code = 0;
200 201
  if (pWal->pWriteIdxTFile != NULL) {
    code = taosCloseFile(&pWal->pWriteIdxTFile);
L
Liu Jicong 已提交
202
    if (code != 0) {
L
Liu Jicong 已提交
203 204
      return -1;
    }
L
Liu Jicong 已提交
205
  }
206 207
  if (pWal->pWriteLogTFile != NULL) {
    code = taosCloseFile(&pWal->pWriteLogTFile);
L
Liu Jicong 已提交
208
    if (code != 0) {
L
Liu Jicong 已提交
209 210
      return -1;
    }
L
Liu Jicong 已提交
211
  }
212
  TdFilePtr pIdxTFile, pLogTFile;
L
Liu Jicong 已提交
213
  // create new file
L
Liu Jicong 已提交
214
  int64_t newFileFirstVersion = pWal->vers.lastVer + 1;
L
Liu Jicong 已提交
215
  char    fnameStr[WAL_FILE_LEN];
L
Liu Jicong 已提交
216
  walBuildIdxName(pWal, newFileFirstVersion, fnameStr);
217
  pIdxTFile = taosOpenFile(fnameStr, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_APPEND);
218
  if (pIdxTFile == NULL) {
L
Liu Jicong 已提交
219
    terrno = TAOS_SYSTEM_ERROR(errno);
L
Liu Jicong 已提交
220 221 222
    return -1;
  }
  walBuildLogName(pWal, newFileFirstVersion, fnameStr);
223
  pLogTFile = taosOpenFile(fnameStr, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_APPEND);
224
  if (pLogTFile == NULL) {
L
Liu Jicong 已提交
225
    terrno = TAOS_SYSTEM_ERROR(errno);
L
Liu Jicong 已提交
226 227 228
    return -1;
  }
  code = walRollFileInfo(pWal);
L
Liu Jicong 已提交
229
  if (code != 0) {
L
Liu Jicong 已提交
230 231
    return -1;
  }
L
Liu Jicong 已提交
232

L
Liu Jicong 已提交
233
  // switch file
234 235
  pWal->pWriteIdxTFile = pIdxTFile;
  pWal->pWriteLogTFile = pLogTFile;
L
Liu Jicong 已提交
236
  pWal->writeCur = taosArrayGetSize(pWal->fileInfoSet) - 1;
L
fix  
Liu Jicong 已提交
237
  ASSERT(pWal->writeCur >= 0);
L
Liu Jicong 已提交
238 239 240 241 242

  pWal->lastRollSeq = walGetSeq();
  return 0;
}

L
Liu Jicong 已提交
243
static int walWriteIndex(SWal *pWal, int64_t ver, int64_t offset) {
L
Liu Jicong 已提交
244
  SWalIdxEntry entry = {.ver = ver, .offset = offset};
245
  int          size = taosWriteFile(pWal->pWriteIdxTFile, &entry, sizeof(SWalIdxEntry));
L
Liu Jicong 已提交
246
  if (size != sizeof(SWalIdxEntry)) {
L
Liu Jicong 已提交
247
    terrno = TAOS_SYSTEM_ERROR(errno);
L
Liu Jicong 已提交
248
    // TODO truncate
L
Liu Jicong 已提交
249
    return -1;
L
Liu Jicong 已提交
250 251 252 253
  }
  return 0;
}

L
Liu Jicong 已提交
254
int64_t walWriteWithSyncInfo(SWal *pWal, int64_t index, tmsg_t msgType, SSyncLogMeta syncMeta, const void *body,
L
Liu Jicong 已提交
255
                             int32_t bodyLen) {
L
Liu Jicong 已提交
256
  if (pWal == NULL) return -1;
L
Liu Jicong 已提交
257
  int code = 0;
陶建辉(Jeff)'s avatar
陶建辉(Jeff) 已提交
258

S
TD-1652  
Shengliang Guan 已提交
259
  // no wal
L
Liu Jicong 已提交
260
  if (pWal->cfg.level == TAOS_WAL_NOLOG) return 0;
L
Liu Jicong 已提交
261

L
Liu Jicong 已提交
262
  if (index == pWal->vers.lastVer + 1) {
L
Liu Jicong 已提交
263
    if (taosArrayGetSize(pWal->fileInfoSet) == 0) {
L
Liu Jicong 已提交
264
      pWal->vers.firstVer = index;
L
Liu Jicong 已提交
265 266
      code = walRoll(pWal);
      ASSERT(code == 0);
L
Liu Jicong 已提交
267
    } else {
L
Liu Jicong 已提交
268
      int64_t passed = walGetSeq() - pWal->lastRollSeq;
L
Liu Jicong 已提交
269
      if (pWal->cfg.rollPeriod != -1 && pWal->cfg.rollPeriod != 0 && passed > pWal->cfg.rollPeriod) {
L
Liu Jicong 已提交
270
        walRoll(pWal);
L
Liu Jicong 已提交
271
      } else if (pWal->cfg.segSize != -1 && pWal->cfg.segSize != 0 && walGetLastFileSize(pWal) > pWal->cfg.segSize) {
L
Liu Jicong 已提交
272 273
        walRoll(pWal);
      }
L
Liu Jicong 已提交
274 275
    }
  } else {
L
Liu Jicong 已提交
276 277
    // reject skip log or rewrite log
    // must truncate explicitly first
L
fix  
Liu Jicong 已提交
278
    terrno = TSDB_CODE_WAL_INVALID_VER;
L
Liu Jicong 已提交
279 280
    return -1;
  }
281
  /*if (!tfValid(pWal->pWriteLogTFile)) return -1;*/
陶建辉(Jeff)'s avatar
陶建辉(Jeff) 已提交
282

L
Liu Jicong 已提交
283 284
  ASSERT(pWal->writeCur >= 0);

wafwerar's avatar
wafwerar 已提交
285
  taosThreadMutexLock(&pWal->mutex);
L
Liu Jicong 已提交
286

287
  if (pWal->pWriteIdxTFile == NULL || pWal->pWriteLogTFile == NULL) {
L
Liu Jicong 已提交
288
    walSetWrite(pWal);
289 290
    taosLSeekFile(pWal->pWriteLogTFile, 0, SEEK_END);
    taosLSeekFile(pWal->pWriteIdxTFile, 0, SEEK_END);
L
Liu Jicong 已提交
291 292
  }

L
Liu Jicong 已提交
293
  pWal->writeHead.head.version = index;
L
Liu Jicong 已提交
294

L
Liu Jicong 已提交
295
  int64_t offset = walGetCurFileOffset(pWal);
L
Liu Jicong 已提交
296
  pWal->writeHead.head.bodyLen = bodyLen;
L
Liu Jicong 已提交
297
  pWal->writeHead.head.msgType = msgType;
L
Liu Jicong 已提交
298

299
  // sync info for sync module
L
Liu Jicong 已提交
300
  pWal->writeHead.head.syncMeta = syncMeta;
L
Liu Jicong 已提交
301

L
Liu Jicong 已提交
302 303
  pWal->writeHead.cksumHead = walCalcHeadCksum(&pWal->writeHead);
  pWal->writeHead.cksumBody = walCalcBodyCksum(body, bodyLen);
304

305
  if (taosWriteFile(pWal->pWriteLogTFile, &pWal->writeHead, sizeof(SWalHead)) != sizeof(SWalHead)) {
306
    // TODO ftruncate
L
fix  
Liu Jicong 已提交
307
    terrno = TAOS_SYSTEM_ERROR(errno);
L
Liu Jicong 已提交
308 309
    wError("vgId:%d, file:%" PRId64 ".log, failed to write since %s", pWal->cfg.vgId, walGetLastFileFirstVer(pWal),
           strerror(errno));
L
fix  
Liu Jicong 已提交
310
    return -1;
J
Jeff Tao 已提交
311
  }
S
TD-1846  
Shengliang Guan 已提交
312

313
  if (taosWriteFile(pWal->pWriteLogTFile, (char *)body, bodyLen) != bodyLen) {
314
    // TODO ftruncate
L
fix  
Liu Jicong 已提交
315
    terrno = TAOS_SYSTEM_ERROR(errno);
L
Liu Jicong 已提交
316 317
    wError("vgId:%d, file:%" PRId64 ".log, failed to write since %s", pWal->cfg.vgId, walGetLastFileFirstVer(pWal),
           strerror(errno));
L
fix  
Liu Jicong 已提交
318
    return -1;
L
Liu Jicong 已提交
319
  }
L
Liu Jicong 已提交
320

L
Liu Jicong 已提交
321
  code = walWriteIndex(pWal, index, offset);
L
Liu Jicong 已提交
322 323
  if (code != 0) {
    // TODO
L
Liu Jicong 已提交
324
    return -1;
L
Liu Jicong 已提交
325
  }
326

L
Liu Jicong 已提交
327
  // set status
L
Liu Jicong 已提交
328
  pWal->vers.lastVer = index;
L
Liu Jicong 已提交
329
  pWal->totSize += sizeof(SWalHead) + bodyLen;
L
Liu Jicong 已提交
330 331
  walGetCurFileInfo(pWal)->lastVer = index;
  walGetCurFileInfo(pWal)->fileSize += sizeof(SWalHead) + bodyLen;
L
Liu Jicong 已提交
332

wafwerar's avatar
wafwerar 已提交
333
  taosThreadMutexUnlock(&pWal->mutex);
陶建辉(Jeff)'s avatar
陶建辉(Jeff) 已提交
334

L
fix  
Liu Jicong 已提交
335
  return 0;
陶建辉(Jeff)'s avatar
陶建辉(Jeff) 已提交
336 337
}

L
Liu Jicong 已提交
338
int64_t walWrite(SWal *pWal, int64_t index, tmsg_t msgType, const void *body, int32_t bodyLen) {
L
Liu Jicong 已提交
339
  SSyncLogMeta syncMeta = {
L
Liu Jicong 已提交
340 341 342 343
      .isWeek = -1,
      .seqNum = UINT64_MAX,
      .term = UINT64_MAX,
  };
L
Liu Jicong 已提交
344
  return walWriteWithSyncInfo(pWal, index, msgType, syncMeta, body, bodyLen);
L
Liu Jicong 已提交
345 346
}

L
Liu Jicong 已提交
347
void walFsync(SWal *pWal, bool forceFsync) {
L
Liu Jicong 已提交
348
  if (forceFsync || (pWal->cfg.level == TAOS_WAL_FSYNC && pWal->cfg.fsyncPeriod == 0)) {
L
Liu Jicong 已提交
349
    wTrace("vgId:%d, fileId:%" PRId64 ".log, do fsync", pWal->cfg.vgId, walGetCurFileFirstVer(pWal));
350
    if (taosFsyncFile(pWal->pWriteLogTFile) < 0) {
L
Liu Jicong 已提交
351 352
      wError("vgId:%d, file:%" PRId64 ".log, fsync failed since %s", pWal->cfg.vgId, walGetCurFileFirstVer(pWal),
             strerror(errno));
353 354
    }
  }
陶建辉(Jeff)'s avatar
陶建辉(Jeff) 已提交
355 356
}

L
Liu Jicong 已提交
357
/*static int walValidateOffset(SWal* pWal, int64_t ver) {*/
L
Liu Jicong 已提交
358 359 360 361 362 363 364
/*int code = 0;*/
/*SWalHead *pHead = NULL;*/
/*code = (int)walRead(pWal, &pHead, ver);*/
/*if(pHead->head.version != ver) {*/
/*return -1;*/
/*}*/
/*return 0;*/
L
Liu Jicong 已提交
365
/*}*/
L
Liu Jicong 已提交
366

L
Liu Jicong 已提交
367
/*static int64_t walGetOffset(SWal* pWal, int64_t ver) {*/
L
Liu Jicong 已提交
368 369 370 371
/*int code = walSeekVer(pWal, ver);*/
/*if(code != 0) {*/
/*return -1;*/
/*}*/
L
Liu Jicong 已提交
372

L
Liu Jicong 已提交
373 374 375 376
/*code = walValidateOffset(pWal, ver);*/
/*if(code != 0) {*/
/*return -1;*/
/*}*/
L
Liu Jicong 已提交
377

L
Liu Jicong 已提交
378
/*return 0;*/
L
Liu Jicong 已提交
379
/*}*/