~drizzle-trunk/drizzle/development : revision 1909.1.1

70

namespace internal

71

{

72

73

static int _my_b_read(register IO_CACHE *info, unsigned char *Buffer, size_t Count);

74

static int _my_b_write(register IO_CACHE *info, const unsigned char *Buffer, size_t Count);

73

static int _my_b_read(register st_io_cache *info, unsigned char *Buffer, size_t Count);

74

static int _my_b_write(register st_io_cache *info, const unsigned char *Buffer, size_t Count);

75

76

/**

77

* @brief

78

* Lock appends for the IO_CACHE if required (need_append_buffer_lock)

78

* Lock appends for the st_io_cache if required (need_append_buffer_lock)

79

*/

80

inline

81

static void lock_append_buffer(register IO_CACHE *, int )

81

static void lock_append_buffer(register st_io_cache *, int )

82

{

83

}

84

85

/**

86

* @brief

87

* Unlock appends for the IO_CACHE if required (need_append_buffer_lock)

87

* Unlock appends for the st_io_cache if required (need_append_buffer_lock)

88

*/

89

inline

90

static void unlock_append_buffer(register IO_CACHE *, int )

90

static void unlock_append_buffer(register st_io_cache *, int )

91

{

92

}

93

114

115

/**

116

* @brief

117

* Setup internal pointers inside IO_CACHE

117

* Setup internal pointers inside st_io_cache

118

*

119

* @details

120

* This is called on automatically on init or reinit of IO_CACHE

121

* It must be called externally if one moves or copies an IO_CACHE object.

120

* This is called on automatically on init or reinit of st_io_cache

121

* It must be called externally if one moves or copies an st_io_cache object.

122

*

123

* @param info Cache handler to setup

124

*/

125

void setup_io_cache(IO_CACHE* info)

125

void st_io_cache::setup_io_cache()

126

{

127

/* Ensure that my_b_tell() and my_b_bytes_in_cache works */

128

if (info->type == WRITE_CACHE)

128

if (type == WRITE_CACHE)

129

{

130

info->current_pos= &info->write_pos;

131

info->current_end= &info->write_end;

130

current_pos= &write_pos;

131

current_end= &write_end;

132

}

133

else

134

{

135

info->current_pos= &info->read_pos;

136

info->current_end= &info->read_end;

135

current_pos= &read_pos;

136

current_end= &read_end;

137

}

138

}

139

140

141

static void

142

init_functions(IO_CACHE* info)

141

void st_io_cache::init_functions()

143

142

{

144

enum cache_type type= info->type;

145

143

switch (type) {

146

144

case READ_NET:

147

145

/*

153

151

*/

154

152

break;

155

153

default:

156

info->read_function = _my_b_read;

157

info->write_function = _my_b_write;

154

read_function = _my_b_read;

155

write_function = _my_b_write;

158

156

}

159

157

160

setup_io_cache(info);

158

setup_io_cache();

161

159

}

162

160

163

161

/**

164

162

* @brief

165

* Initialize an IO_CACHE object

163

* Initialize an st_io_cache object

166

164

*

167

* @param info Cache handler to initialize

168

165

* @param file File that should be associated with the handler

169

166

* If == -1 then real_open_cached_file() will be called when it's time to open file.

170

167

* @param cachesize Size of buffer to allocate for read/write

178

175

* @retval 0 success

179

176

* @retval # error

180

177

*/

181

int init_io_cache(IO_CACHE *info, int file, size_t cachesize,

182

enum cache_type type, my_off_t seek_offset,

183

bool use_async_io, myf cache_myflags)

178

int st_io_cache::init_io_cache(int file_arg, size_t cachesize,

179

enum cache_type type_arg, my_off_t seek_offset,

180

bool use_async_io, myf cache_myflags)

184

181

{

185

182

size_t min_cache;

186

183

off_t pos;

187

my_off_t end_of_file= ~(my_off_t) 0;

184

my_off_t end_of_file_local= ~(my_off_t) 0;

188

185

189

info->file= file;

190

info->type= TYPE_NOT_SET; /* Don't set it until mutex are created */

191

info->pos_in_file= seek_offset;

192

info->pre_close = info->pre_read = info->post_read = 0;

193

info->arg = 0;

194

info->alloced_buffer = 0;

195

info->buffer=0;

196

info->seek_not_done= 0;

186

file= file_arg;

187

type= TYPE_NOT_SET; /* Don't set it until mutex are created */

188

pos_in_file= seek_offset;

189

pre_close = pre_read = post_read = 0;

190

arg = 0;

191

alloced_buffer = 0;

192

buffer=0;

193

seek_not_done= 0;

197

194

198

195

if (file >= 0)

199

196

{

204

201

This kind of object doesn't support seek() or tell(). Don't set a

205

202

flag that will make us again try to seek() later and fail.

206

203

*/

207

info->seek_not_done= 0;

204

seek_not_done= 0;

208

205

/*

209

206

Additionally, if we're supposed to start somewhere other than the

210

207

the beginning of whatever this file is, then somebody made a bad

213

210

assert(seek_offset == 0);

214

211

}

215

212

else

216

info->seek_not_done= test(seek_offset != (my_off_t)pos);

213

seek_not_done= test(seek_offset != (my_off_t)pos);

217

214

}

218

215

219

216

if (!cachesize && !(cachesize= my_default_record_cache_size))

220

return(1); /* No cache requested */

217

return 1; /* No cache requested */

221

218

min_cache=use_async_io ? IO_SIZE*4 : IO_SIZE*2;

222

if (type == READ_CACHE)

219

if (type_arg == READ_CACHE)

223

220

{ /* Assume file isn't growing */

224

221

if (!(cache_myflags & MY_DONT_CHECK_FILESIZE))

225

222

{

226

223

/* Calculate end of file to avoid allocating oversized buffers */

227

end_of_file=lseek(file,0L,SEEK_END);

224

end_of_file_local=lseek(file,0L,SEEK_END);

228

225

/* Need to reset seek_not_done now that we just did a seek. */

229

info->seek_not_done= end_of_file == seek_offset ? 0 : 1;

230

if (end_of_file < seek_offset)

231

end_of_file=seek_offset;

226

seek_not_done= end_of_file_local == seek_offset ? 0 : 1;

227

if (end_of_file_local < seek_offset)

228

end_of_file_local=seek_offset;

232

229

/* Trim cache size if the file is very small */

233

if ((my_off_t) cachesize > end_of_file-seek_offset+IO_SIZE*2-1)

230

if ((my_off_t) cachesize > end_of_file_local-seek_offset+IO_SIZE*2-1)

234

231

{

235

cachesize= (size_t) (end_of_file-seek_offset)+IO_SIZE*2-1;

232

cachesize= (size_t) (end_of_file_local-seek_offset)+IO_SIZE*2-1;

236

233

use_async_io=0; /* No need to use async */

237

234

}

238

235

}

239

236

}

240

237

cache_myflags &= ~MY_DONT_CHECK_FILESIZE;

241

if (type != READ_NET && type != WRITE_NET)

238

if (type_arg != READ_NET && type_arg != WRITE_NET)

242

239

{

243

240

/* Retry allocating memory in smaller blocks until we get one */

244

241

cachesize= ((cachesize + min_cache-1) & ~(min_cache-1));

248

245

if (cachesize < min_cache)

249

246

cachesize = min_cache;

250

247

buffer_block= cachesize;

251

if ((type == READ_CACHE) and (not global_read_buffer.add(buffer_block)))

248

if ((type_arg == READ_CACHE) and (not global_read_buffer.add(buffer_block)))

252

249

{

253

250

my_error(ER_OUT_OF_GLOBAL_READMEMORY, MYF(ME_ERROR+ME_WAITTANG));

254

251

return 2;

255

252

}

256

253

257

if ((info->buffer=

254

if ((buffer=

258

255

(unsigned char*) malloc(buffer_block)) != 0)

259

256

{

260

info->write_buffer=info->buffer;

261

info->alloced_buffer= true;

257

write_buffer=buffer;

258

alloced_buffer= true;

262

259

break; /* Enough memory found */

263

260

}

264

261

if (cachesize == min_cache)

265

262

{

266

if (type == READ_CACHE)

263

if (type_arg == READ_CACHE)

267

264

global_read_buffer.sub(buffer_block);

268

return(2); /* Can't alloc cache */

265

return 2; /* Can't alloc cache */

269

266

}

270

267

/* Try with less memory */

271

if (type == READ_CACHE)

268

if (type_arg == READ_CACHE)

272

269

global_read_buffer.sub(buffer_block);

273

270

cachesize= (cachesize*3/4 & ~(min_cache-1));

274

271

}

275

272

}

276

273

277

info->read_length=info->buffer_length=cachesize;

278

info->myflags=cache_myflags & ~(MY_NABP | MY_FNABP);

279

info->request_pos= info->read_pos= info->write_pos = info->buffer;

274

read_length=buffer_length=cachesize;

275

myflags=cache_myflags & ~(MY_NABP | MY_FNABP);

276

request_pos= read_pos= write_pos = buffer;

280

277

281

if (type == WRITE_CACHE)

282

info->write_end=

283

info->buffer+info->buffer_length- (seek_offset & (IO_SIZE-1));

278

if (type_arg == WRITE_CACHE)

279

write_end=

280

buffer+buffer_length- (seek_offset & (IO_SIZE-1));

284

281

else

285

info->read_end=info->buffer; /* Nothing in cache */

282

read_end=buffer; /* Nothing in cache */

286

283

287

284

/* End_of_file may be changed by user later */

288

info->end_of_file= end_of_file;

289

info->error=0;

290

info->type= type;

291

init_functions(info);

285

end_of_file= end_of_file_local;

286

error= 0;

287

type= type_arg;

288

init_functions();

292

289

#ifdef HAVE_AIOWAIT

293

290

if (use_async_io && ! my_disable_async_io)

294

291

{

295

info->read_length/=2;

296

info->read_function=_my_b_async_read;

292

read_length/=2;

293

read_function=_my_b_async_read;

297

294

}

298

info->inited=info->aio_result.pending=0;

295

inited= aio_result.pending= 0;

299

296

#endif

300

return(0);

297

return 0;

301

298

} /* init_io_cache */

302

299

303

300

/* Wait until current request is ready */

322

319

break;

323

320

}

324

321

}

325

return;

326

322

}

327

323

#endif

328

324

336

332

* If we are doing a reinit of a cache where we have the start of the file

337

333

* in the cache, we are reusing this memory without flushing it to disk.

338

334

*/

339

bool reinit_io_cache(IO_CACHE *info, enum cache_type type,

340

my_off_t seek_offset,

341

bool use_async_io,

342

bool clear_cache)

335

bool st_io_cache::reinit_io_cache(enum cache_type type_arg,

336

my_off_t seek_offset,

337

bool use_async_io,

338

bool clear_cache)

343

339

{

344

340

/* One can't do reinit with the following types */

345

assert(type != READ_NET && info->type != READ_NET &&

346

type != WRITE_NET && info->type != WRITE_NET);

341

assert(type_arg != READ_NET && type != READ_NET &&

342

type_arg != WRITE_NET && type != WRITE_NET);

347

343

348

344

/* If the whole file is in memory, avoid flushing to disk */

349

345

if (! clear_cache &&

350

seek_offset >= info->pos_in_file &&

351

seek_offset <= my_b_tell(info))

346

seek_offset >= pos_in_file &&

347

seek_offset <= my_b_tell(this))

352

348

{

353

349

/* Reuse current buffer without flushing it to disk */

354

350

unsigned char *pos;

355

if (info->type == WRITE_CACHE && type == READ_CACHE)

351

if (type == WRITE_CACHE && type_arg == READ_CACHE)

356

352

{

357

info->read_end=info->write_pos;

358

info->end_of_file=my_b_tell(info);

353

read_end=write_pos;

354

end_of_file=my_b_tell(this);

359

355

/*

360

356

Trigger a new seek only if we have a valid

361

357

file handle.

362

358

*/

363

info->seek_not_done= (info->file != -1);

359

seek_not_done= (file != -1);

364

360

}

365

else if (type == WRITE_CACHE)

361

else if (type_arg == WRITE_CACHE)

366

362

{

367

if (info->type == READ_CACHE)

363

if (type == READ_CACHE)

368

364

{

369

info->write_end=info->write_buffer+info->buffer_length;

370

info->seek_not_done=1;

365

write_end=write_buffer+buffer_length;

366

seek_not_done=1;

371

367

}

372

info->end_of_file = ~(my_off_t) 0;

368

end_of_file = ~(my_off_t) 0;

373

369

}

374

pos=info->request_pos+(seek_offset-info->pos_in_file);

375

if (type == WRITE_CACHE)

376

info->write_pos=pos;

370

pos=request_pos+(seek_offset-pos_in_file);

371

if (type_arg == WRITE_CACHE)

372

write_pos=pos;

377

373

else

378

info->read_pos= pos;

374

read_pos= pos;

379

375

#ifdef HAVE_AIOWAIT

380

my_aiowait(&info->aio_result); /* Wait for outstanding req */

376

my_aiowait(&aio_result); /* Wait for outstanding req */

381

377

#endif

382

378

}

383

379

else

386

382

If we change from WRITE_CACHE to READ_CACHE, assume that everything

387

383

after the current positions should be ignored

388

384

*/

389

if (info->type == WRITE_CACHE && type == READ_CACHE)

390

info->end_of_file=my_b_tell(info);

385

if (type == WRITE_CACHE && type_arg == READ_CACHE)

386

end_of_file=my_b_tell(this);

391

387

/* flush cache if we want to reuse it */

392

if (!clear_cache && my_b_flush_io_cache(info,1))

393

return(1);

394

info->pos_in_file=seek_offset;

388

if (!clear_cache && my_b_flush_io_cache(this, 1))

389

return 1;

390

pos_in_file=seek_offset;

395

391

/* Better to do always do a seek */

396

info->seek_not_done=1;

397

info->request_pos=info->read_pos=info->write_pos=info->buffer;

398

if (type == READ_CACHE)

392

seek_not_done=1;

393

request_pos=read_pos=write_pos=buffer;

394

if (type_arg == READ_CACHE)

399

395

{

400

info->read_end=info->buffer; /* Nothing in cache */

396

read_end=buffer; /* Nothing in cache */

401

397

}

402

398

else

403

399

{

404

info->write_end=(info->buffer + info->buffer_length -

400

write_end=(buffer + buffer_length -

405

401

(seek_offset & (IO_SIZE-1)));

406

info->end_of_file= ~(my_off_t) 0;

402

end_of_file= ~(my_off_t) 0;

407

403

}

408

404

}

409

info->type=type;

410

info->error=0;

411

init_functions(info);

405

type= type_arg;

406

error=0;

407

init_functions();

412

408

413

409

#ifdef HAVE_AIOWAIT

414

410

if (use_async_io && ! my_disable_async_io &&

415

((uint32_t) info->buffer_length <

416

(uint32_t) (info->end_of_file - seek_offset)))

411

((uint32_t) buffer_length <

412

(uint32_t) (end_of_file - seek_offset)))

417

413

{

418

info->read_length=info->buffer_length/2;

419

info->read_function=_my_b_async_read;

414

read_length=buffer_length/2;

415

read_function=_my_b_async_read;

420

416

}

421

info->inited=0;

417

inited= 0;

422

418

#else

423

419

(void)use_async_io;

424

420

#endif

425

return(0);

421

return 0;

426

422

} /* reinit_io_cache */

427

423

428

424

/**

439

435

* types than my_off_t unless you can be sure that their value fits.

440

436

* Same applies to differences of file offsets.

441

437

*

442

* @param info IO_CACHE pointer @param Buffer Buffer to retrieve count bytes

438

* @param info st_io_cache pointer @param Buffer Buffer to retrieve count bytes

443

439

* from file @param Count Number of bytes to read into Buffer

444

440

*

445

441

* @retval 0 We succeeded in reading all data

446

442

* @retval 1 Error: can't read requested characters

447

443

*/

448

static int _my_b_read(register IO_CACHE *info, unsigned char *Buffer, size_t Count)

444

static int _my_b_read(register st_io_cache *info, unsigned char *Buffer, size_t Count)

449

445

{

450

446

size_t length,diff_length,left_length, max_length;

451

447

my_off_t pos_in_file;

462

458

pos_in_file=info->pos_in_file+ (size_t) (info->read_end - info->buffer);

463

459

464

460

/*

465

Whenever a function which operates on IO_CACHE flushes/writes

466

some part of the IO_CACHE to disk it will set the property

461

Whenever a function which operates on st_io_cache flushes/writes

462

some part of the st_io_cache to disk it will set the property

467

463

"seek_not_done" to indicate this to other functions operating

468

on the IO_CACHE.

464

on the st_io_cache.

469

465

*/

470

466

if (info->seek_not_done)

471

467

{

547

543

548

544

/**

549

545

* @brief

550

* Read from the IO_CACHE into a buffer and feed asynchronously from disk when needed.

546

* Read from the st_io_cache into a buffer and feed asynchronously from disk when needed.

551

547

*

552

* @param info IO_CACHE pointer

548

* @param info st_io_cache pointer

553

549

* @param Buffer Buffer to retrieve count bytes from file

554

550

* @param Count Number of bytes to read into Buffer

555

551

*

556

552

* @retval -1 An error has occurred; errno is set.

557

553

* @retval 0 Success

558

* @retval 1 An error has occurred; IO_CACHE to error state.

554

* @retval 1 An error has occurred; st_io_cache to error state.

559

555

*/

560

int _my_b_async_read(register IO_CACHE *info, unsigned char *Buffer, size_t Count)

556

int _my_b_async_read(register st_io_cache *info, unsigned char *Buffer, size_t Count)

561

557

{

562

558

size_t length,read_length,diff_length,left_length,use_length,org_Count;

563

559

size_t max_length;

724

720

* @brief

725

721

* Read one byte when buffer is empty

726

722

*/

727

int _my_b_get(IO_CACHE *info)

723

int _my_b_get(st_io_cache *info)

728

724

{

729

725

unsigned char buff;

730

726

IO_CACHE_CALLBACK pre_read,post_read;

739

735

740

736

/**

741

737

* @brief

742

* Write a byte buffer to IO_CACHE and flush to disk if IO_CACHE is full.

738

* Write a byte buffer to st_io_cache and flush to disk if st_io_cache is full.

743

739

*

744

740

* @retval -1 On error; errno contains error code.

745

741

* @retval 0 On success

746

742

* @retval 1 On error on write

747

743

*/

748

int _my_b_write(register IO_CACHE *info, const unsigned char *Buffer, size_t Count)

744

int _my_b_write(register st_io_cache *info, const unsigned char *Buffer, size_t Count)

749

745

{

750

746

size_t rest_length,length;

751

747

769

765

if (info->seek_not_done)

770

766

{

771

767

/*

772

Whenever a function which operates on IO_CACHE flushes/writes

773

some part of the IO_CACHE to disk it will set the property

768

Whenever a function which operates on st_io_cache flushes/writes

769

some part of the st_io_cache to disk it will set the property

774

770

"seek_not_done" to indicate this to other functions operating

775

on the IO_CACHE.

771

on the st_io_cache.

776

772

*/

777

773

if (lseek(info->file,info->pos_in_file,SEEK_SET))

778

774

{

799

795

* As all write calls to the data goes through the cache,

800

796

* we will never get a seek over the end of the buffer.

801

797

*/

802

int my_block_write(register IO_CACHE *info, const unsigned char *Buffer, size_t Count,

798

int my_block_write(register st_io_cache *info, const unsigned char *Buffer, size_t Count,

803

799

my_off_t pos)

804

800

{

805

801

size_t length;

846

842

* @brief

847

843

* Flush write cache

848

844

*/

849

int my_b_flush_io_cache(IO_CACHE *info, int need_append_buffer_lock)

845

int my_b_flush_io_cache(st_io_cache *info, int need_append_buffer_lock)

850

846

{

851

847

size_t length;

852

848

bool append_cache= false;

856

852

{

857

853

if (info->file == -1)

858

854

{

859

if (real_open_cached_file(info))

855

if (info->real_open_cached_file())

860

856

return((info->error= -1));

861

857

}

862

858

lock_append_buffer(info, need_append_buffer_lock);

917

913

918

914

/**

919

915

* @brief

920

* Free an IO_CACHE object

916

* Free an st_io_cache object

921

917

*

922

918

* @detail

923

919

* It's currently safe to call this if one has called init_io_cache()

924

920

* on the 'info' object, even if init_io_cache() failed.

925

921

* This function is also safe to call twice with the same handle.

926

922

*

927

* @param info IO_CACHE Handle to free

923

* @param info st_io_cache Handle to free

928

924

*

929

925

* @retval 0 ok

930

926

* @retval # Error

931

927

*/

932

int end_io_cache(IO_CACHE *info)

928

int st_io_cache::end_io_cache()

933

929

{

934

int error=0;

935

IO_CACHE_CALLBACK pre_close;

936

937

if ((pre_close=info->pre_close))

938

{

939

(*pre_close)(info);

940

info->pre_close= 0;

941

}

942

if (info->alloced_buffer)

943

{

944

if (info->type == READ_CACHE)

945

global_read_buffer.sub(info->buffer_length);

946

info->alloced_buffer=0;

947

if (info->file != -1) /* File doesn't exist */

948

error= my_b_flush_io_cache(info,1);

949

free((unsigned char*) info->buffer);

950

info->buffer=info->read_pos=(unsigned char*) 0;

951

}

952

953

return(error);

930

int _error=0;

931

932

if (pre_close)

933

{

934

(*pre_close)(this);

935

pre_close= 0;

936

}

937

if (alloced_buffer)

938

{

939

if (type == READ_CACHE)

940

global_read_buffer.sub(buffer_length);

941

alloced_buffer=0;

942

if (file != -1) /* File doesn't exist */

943

_error= my_b_flush_io_cache(this, 1);

944

free((unsigned char*) buffer);

945

buffer=read_pos=(unsigned char*) 0;

946

}

947

948

return _error;

954

949

} /* end_io_cache */

955

950

956

951

} /* namespace internal */