walRead.c 6.1 KB
Newer Older
H
refact  
Hongze Cheng 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13
/*
 * 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/>.
S
Shengliang Guan 已提交
14 15
 */

L
Liu Jicong 已提交
16
#include "walInt.h"
L
Liu Jicong 已提交
17
#include "taoserror.h"
S
Shengliang Guan 已提交
18

L
Liu Jicong 已提交
19
SWalReadHandle *walOpenReadHandle(SWal *pWal) {
L
Liu Jicong 已提交
20
  SWalReadHandle *pRead = malloc(sizeof(SWalReadHandle));
L
Liu Jicong 已提交
21
  if (pRead == NULL) {
22
    terrno = TSDB_CODE_OUT_OF_MEMORY;
L
Liu Jicong 已提交
23 24
    return NULL;
  }
25

L
Liu Jicong 已提交
26
  pRead->pWal = pWal;
27 28
  pRead->pReadIdxTFile = NULL;
  pRead->pReadLogTFile = NULL;
L
Liu Jicong 已提交
29 30 31 32 33
  pRead->curVersion = -1;
  pRead->curFileFirstVer = -1;
  pRead->capacity = 0;
  pRead->status = 0;
  pRead->pHead = malloc(sizeof(SWalHead));
L
Liu Jicong 已提交
34
  if (pRead->pHead == NULL) {
L
Liu Jicong 已提交
35
    terrno = TSDB_CODE_WAL_OUT_OF_MEMORY;
L
Liu Jicong 已提交
36 37 38 39
    free(pRead);
    return NULL;
  }
  return pRead;
L
Liu Jicong 已提交
40 41 42
}

void walCloseReadHandle(SWalReadHandle *pRead) {
43 44
  taosCloseFile(&pRead->pReadIdxTFile);
  taosCloseFile(&pRead->pReadLogTFile);
L
Liu Jicong 已提交
45
  tfree(pRead->pHead);
L
Liu Jicong 已提交
46 47 48
  free(pRead);
}

L
Liu Jicong 已提交
49
int32_t walRegisterRead(SWalReadHandle *pRead, int64_t ver) { return 0; }
L
Liu Jicong 已提交
50 51 52 53

static int32_t walReadSeekFilePos(SWalReadHandle *pRead, int64_t fileFirstVer, int64_t ver) {
  int code = 0;

54 55
  TdFilePtr pIdxTFile = pRead->pReadIdxTFile;
  TdFilePtr pLogTFile = pRead->pReadLogTFile;
L
Liu Jicong 已提交
56 57

  // seek position
L
Liu Jicong 已提交
58
  int64_t offset = (ver - fileFirstVer) * sizeof(SWalIdxEntry);
59
  code = taosLSeekFile(pIdxTFile, offset, SEEK_SET);
L
Liu Jicong 已提交
60
  if (code < 0) {
L
Liu Jicong 已提交
61
    terrno = TAOS_SYSTEM_ERROR(errno);
L
Liu Jicong 已提交
62 63
    return -1;
  }
L
Liu Jicong 已提交
64
  SWalIdxEntry entry;
65
  if (taosReadFile(pIdxTFile, &entry, sizeof(SWalIdxEntry)) != sizeof(SWalIdxEntry)) {
L
Liu Jicong 已提交
66
    terrno = TSDB_CODE_WAL_FILE_CORRUPTED;
L
Liu Jicong 已提交
67 68
    return -1;
  }
L
Liu Jicong 已提交
69
  // TODO:deserialize
L
Liu Jicong 已提交
70
  ASSERT(entry.ver == ver);
71
  code = taosLSeekFile(pLogTFile, entry.offset, SEEK_SET);
L
Liu Jicong 已提交
72
  if (code < 0) {
L
Liu Jicong 已提交
73
    terrno = TAOS_SYSTEM_ERROR(errno);
L
Liu Jicong 已提交
74 75 76 77 78 79 80 81
    return -1;
  }
  return code;
}

static int32_t walReadChangeFile(SWalReadHandle *pRead, int64_t fileFirstVer) {
  char fnameStr[WAL_FILE_LEN];

82 83
  taosCloseFile(&pRead->pReadIdxTFile);
  taosCloseFile(&pRead->pReadLogTFile);
L
Liu Jicong 已提交
84 85

  walBuildLogName(pRead->pWal, fileFirstVer, fnameStr);
86 87
  TdFilePtr pLogTFile = taosOpenFile(fnameStr, TD_FILE_READ);
  if (pLogTFile == NULL) {
L
Liu Jicong 已提交
88
    terrno = TAOS_SYSTEM_ERROR(errno);
L
Liu Jicong 已提交
89 90 91 92
    return -1;
  }

  walBuildIdxName(pRead->pWal, fileFirstVer, fnameStr);
93 94
  TdFilePtr pIdxTFile = taosOpenFile(fnameStr, TD_FILE_READ);
  if (pIdxTFile == NULL) {
L
Liu Jicong 已提交
95 96 97
    return -1;
  }

98 99
  pRead->pReadLogTFile = pLogTFile;
  pRead->pReadIdxTFile = pIdxTFile;
L
Liu Jicong 已提交
100 101 102 103
  return 0;
}

static int32_t walReadSeekVer(SWalReadHandle *pRead, int64_t ver) {
L
Liu Jicong 已提交
104
  int   code;
L
Liu Jicong 已提交
105
  SWal *pWal = pRead->pWal;
L
Liu Jicong 已提交
106
  if (ver == pRead->curVersion) {
L
Liu Jicong 已提交
107 108
    return 0;
  }
L
Liu Jicong 已提交
109
  if (ver > pWal->vers.lastVer || ver < pWal->vers.firstVer) {
L
Liu Jicong 已提交
110
    terrno = TSDB_CODE_WAL_INVALID_VER;
L
Liu Jicong 已提交
111 112
    return -1;
  }
L
Liu Jicong 已提交
113
  if (ver < pWal->vers.snapshotVer) {
L
Liu Jicong 已提交
114 115
  }

L
Liu Jicong 已提交
116
  SWalFileInfo tmpInfo;
L
Liu Jicong 已提交
117
  tmpInfo.firstVer = ver;
L
Liu Jicong 已提交
118
  // bsearch in fileSet
L
Liu Jicong 已提交
119
  SWalFileInfo *pRet = taosArraySearch(pWal->fileInfoSet, &tmpInfo, compareWalFileInfo, TD_LE);
L
Liu Jicong 已提交
120
  ASSERT(pRet != NULL);
L
Liu Jicong 已提交
121
  if (pRead->curFileFirstVer != pRet->firstVer) {
L
Liu Jicong 已提交
122
    code = walReadChangeFile(pRead, pRet->firstVer);
L
Liu Jicong 已提交
123
    if (code < 0) {
L
Liu Jicong 已提交
124 125 126 127 128
      return -1;
    }
  }

  code = walReadSeekFilePos(pRead, pRet->firstVer, ver);
L
Liu Jicong 已提交
129
  if (code < 0) {
L
Liu Jicong 已提交
130 131 132
    return -1;
  }
  pRead->curVersion = ver;
L
Liu Jicong 已提交
133

L
Liu Jicong 已提交
134 135 136 137 138
  return 0;
}

int32_t walReadWithHandle(SWalReadHandle *pRead, int64_t ver) {
  int code;
L
Liu Jicong 已提交
139 140 141
  // TODO: check wal life
  if (pRead->curVersion != ver) {
    code = walReadSeekVer(pRead, ver);
L
Liu Jicong 已提交
142
    if (code < 0) {
L
Liu Jicong 已提交
143 144
      return -1;
    }
L
Liu Jicong 已提交
145 146
  }

147
  if (!taosValidFile(pRead->pReadLogTFile)) return -1;
L
Liu Jicong 已提交
148

149
  code = taosReadFile(pRead->pReadLogTFile, pRead->pHead, sizeof(SWalHead));
L
Liu Jicong 已提交
150
  if (code != sizeof(SWalHead)) {
L
Liu Jicong 已提交
151 152
    return -1;
  }
L
Liu Jicong 已提交
153
  code = walValidHeadCksum(pRead->pHead);
L
Liu Jicong 已提交
154
  if (code != 0) {
L
Liu Jicong 已提交
155
    terrno = TSDB_CODE_WAL_FILE_CORRUPTED;
L
Liu Jicong 已提交
156 157
    return -1;
  }
L
Liu Jicong 已提交
158 159 160
  if (pRead->capacity < pRead->pHead->head.len) {
    void *ptr = realloc(pRead->pHead, sizeof(SWalHead) + pRead->pHead->head.len);
    if (ptr == NULL) {
L
Liu Jicong 已提交
161
      terrno = TSDB_CODE_WAL_OUT_OF_MEMORY;
L
Liu Jicong 已提交
162 163
      return -1;
    }
L
Liu Jicong 已提交
164 165
    pRead->pHead = ptr;
    pRead->capacity = pRead->pHead->head.len;
L
Liu Jicong 已提交
166
  }
167
  if (pRead->pHead->head.len != taosReadFile(pRead->pReadLogTFile, pRead->pHead->head.body, pRead->pHead->head.len)) {
L
Liu Jicong 已提交
168 169
    return -1;
  }
L
Liu Jicong 已提交
170

L
Liu Jicong 已提交
171
  if (pRead->pHead->head.version != ver) {
L
Liu Jicong 已提交
172
    wError("unexpected wal log version: %ld, read request version:%ld", pRead->pHead->head.version, ver);
L
Liu Jicong 已提交
173 174 175 176
    pRead->curVersion = -1;
    terrno = TSDB_CODE_WAL_FILE_CORRUPTED;
    return -1;
  }
L
Liu Jicong 已提交
177

L
Liu Jicong 已提交
178
  code = walValidBodyCksum(pRead->pHead);
L
Liu Jicong 已提交
179
  if (code != 0) {
L
Liu Jicong 已提交
180
    wError("unexpected wal log version: checksum not passed");
L
Liu Jicong 已提交
181
    pRead->curVersion = -1;
L
Liu Jicong 已提交
182
    terrno = TSDB_CODE_WAL_FILE_CORRUPTED;
L
Liu Jicong 已提交
183 184
    return -1;
  }
L
Liu Jicong 已提交
185
  pRead->curVersion++;
L
Liu Jicong 已提交
186 187 188 189

  return 0;
}

L
Liu Jicong 已提交
190
#if 0
L
Liu Jicong 已提交
191
int32_t walRead(SWal *pWal, SWalHead **ppHead, int64_t ver) {
L
Liu Jicong 已提交
192 193
  int code;
  code = walSeekVer(pWal, ver);
L
Liu Jicong 已提交
194
  if (code != 0) {
L
Liu Jicong 已提交
195 196
    return code;
  }
L
Liu Jicong 已提交
197 198 199
  if (*ppHead == NULL) {
    void *ptr = realloc(*ppHead, sizeof(SWalHead));
    if (ptr == NULL) {
L
Liu Jicong 已提交
200 201 202 203
      return -1;
    }
    *ppHead = ptr;
  }
204
  if (tfRead(pWal->pWriteLogTFile, *ppHead, sizeof(SWalHead)) != sizeof(SWalHead)) {
L
Liu Jicong 已提交
205 206
    return -1;
  }
L
Liu Jicong 已提交
207 208
  // TODO: endian compatibility processing after read
  if (walValidHeadCksum(*ppHead) != 0) {
L
Liu Jicong 已提交
209 210
    return -1;
  }
L
Liu Jicong 已提交
211 212
  void *ptr = realloc(*ppHead, sizeof(SWalHead) + (*ppHead)->head.len);
  if (ptr == NULL) {
L
Liu Jicong 已提交
213 214 215 216
    free(*ppHead);
    *ppHead = NULL;
    return -1;
  }
217
  if (tfRead(pWal->pWriteLogTFile, (*ppHead)->head.body, (*ppHead)->head.len) != (*ppHead)->head.len) {
L
Liu Jicong 已提交
218 219
    return -1;
  }
L
Liu Jicong 已提交
220 221
  // TODO: endian compatibility processing after read
  if (walValidBodyCksum(*ppHead) != 0) {
L
Liu Jicong 已提交
222 223
    return -1;
  }
L
Liu Jicong 已提交
224

L
Liu Jicong 已提交
225 226
  return 0;
}
S
Shengliang Guan 已提交
227

L
Liu Jicong 已提交
228 229 230 231
int32_t walReadWithFp(SWal *pWal, FWalWrite writeFp, int64_t verStart, int32_t readNum) {
return 0;
}
#endif