~drizzle-trunk/drizzle/development : revision 51.1.42

31

32

Relay_log_info::Relay_log_info()

33

:Slave_reporting_capability("SQL"),

34

no_storage(FALSE), replicate_same_server_id(::replicate_same_server_id),

34

no_storage(false), replicate_same_server_id(::replicate_same_server_id),

35

info_fd(-1), cur_log_fd(-1), save_temporary_tables(0),

36

#if defined(HAVE_purify) && HAVE_purify

37

is_fake(FALSE),

37

is_fake(false),

38

#endif

39

cur_log_old_open_count(0), group_relay_log_pos(0), event_relay_log_pos(0),

40

group_master_log_pos(0), log_space_total(0), ignore_log_space_limit(0),

45

tables_to_lock(0), tables_to_lock_count(0),

46

last_event_start_time(0), m_flags(0)

47

{

48

DBUG_ENTER("Relay_log_info::Relay_log_info");

49

50

48

group_relay_log_name[0]= event_relay_log_name[0]=

51

49

group_master_log_name[0]= 0;

52

50

until_log_name[0]= ign_master_log_name_end[0]= 0;

61

59

pthread_cond_init(&stop_cond, NULL);

62

60

pthread_cond_init(&log_space_cond, NULL);

63

61

relay_log.init_pthread_objects();

64

DBUG_VOID_RETURN;

62

return;

65

63

}

66

64

67

65

68

66

Relay_log_info::~Relay_log_info()

69

67

{

70

DBUG_ENTER("Relay_log_info::~Relay_log_info");

71

72

68

pthread_mutex_destroy(&run_lock);

73

69

pthread_mutex_destroy(&data_lock);

74

70

pthread_mutex_destroy(&log_space_lock);

77

73

pthread_cond_destroy(&stop_cond);

78

74

pthread_cond_destroy(&log_space_cond);

79

75

relay_log.cleanup();

80

DBUG_VOID_RETURN;

76

return;

81

77

}

82

78

83

79

88

84

int info_fd;

89

85

const char* msg = 0;

90

86

int error = 0;

91

DBUG_ENTER("init_relay_log_info");

92

DBUG_ASSERT(!rli->no_storage); // Don't init if there is no storage

87

assert(!rli->no_storage); // Don't init if there is no storage

93

88

94

89

if (rli->inited) // Set if this function called

95

DBUG_RETURN(0);

90

return(0);

96

91

fn_format(fname, info_fname, mysql_data_home, "", 4+32);

97

92

pthread_mutex_lock(&rli->data_lock);

98

93

info_fd = rli->info_fd;

153

148

{

154

149

pthread_mutex_unlock(&rli->data_lock);

155

150

sql_print_error("Failed in open_log() called from init_relay_log_info()");

156

DBUG_RETURN(1);

151

return(1);

157

152

}

158

153

}

159

154

221

216

rli->info_fd= -1;

222

217

rli->relay_log.close(LOG_CLOSE_INDEX | LOG_CLOSE_STOP_EVENT);

223

218

pthread_mutex_unlock(&rli->data_lock);

224

DBUG_RETURN(1);

219

return(1);

225

220

}

226

221

}

227

222

258

253

goto err;

259

254

}

260

255

}

261

262

#ifndef DBUG_OFF

263

{

264

char llbuf1[22], llbuf2[22];

265

DBUG_PRINT("info", ("my_b_tell(rli->cur_log)=%s rli->event_relay_log_pos=%s",

266

llstr(my_b_tell(rli->cur_log),llbuf1),

267

llstr(rli->event_relay_log_pos,llbuf2)));

268

DBUG_ASSERT(rli->event_relay_log_pos >= BIN_LOG_HEADER_SIZE);

269

DBUG_ASSERT(my_b_tell(rli->cur_log) == rli->event_relay_log_pos);

270

}

271

#endif

256

char llbuf1[22], llbuf2[22];

257

assert(rli->event_relay_log_pos >= BIN_LOG_HEADER_SIZE);

258

assert(my_b_tell(rli->cur_log) == rli->event_relay_log_pos);

272

259

273

260

/*

274

261

Now change the cache from READ to WRITE - must do this

284

271

}

285

272

rli->inited= 1;

286

273

pthread_mutex_unlock(&rli->data_lock);

287

DBUG_RETURN(error);

274

return(error);

288

275

289

276

err:

290

277

sql_print_error(msg);

294

281

rli->info_fd= -1;

295

282

rli->relay_log.close(LOG_CLOSE_INDEX | LOG_CLOSE_STOP_EVENT);

296

283

pthread_mutex_unlock(&rli->data_lock);

297

DBUG_RETURN(1);

284

return(1);

298

285

}

299

286

300

287

301

288

static inline int add_relay_log(Relay_log_info* rli,LOG_INFO* linfo)

302

289

{

303

290

struct stat s;

304

DBUG_ENTER("add_relay_log");

305

291

if (stat(linfo->log_file_name,&s))

306

292

{

307

293

sql_print_error("log %s listed in the index, but failed to stat",

308

294

linfo->log_file_name);

309

DBUG_RETURN(1);

295

return(1);

310

296

}

311

297

rli->log_space_total += s.st_size;

312

#ifndef DBUG_OFF

313

char buf[22];

314

DBUG_PRINT("info",("log_space_total: %s", llstr(rli->log_space_total,buf)));

315

#endif

316

DBUG_RETURN(0);

298

return(0);

317

299

}

318

300

319

301

320

302

static int count_relay_log_space(Relay_log_info* rli)

321

303

{

322

304

LOG_INFO linfo;

323

DBUG_ENTER("count_relay_log_space");

324

305

rli->log_space_total= 0;

325

306

if (rli->relay_log.find_log_pos(&linfo, NullS, 1))

326

307

{

327

308

sql_print_error("Could not find first log while counting relay log space");

328

DBUG_RETURN(1);

309

return(1);

329

310

}

330

311

do

331

312

{

332

313

if (add_relay_log(rli,&linfo))

333

DBUG_RETURN(1);

314

return(1);

334

315

} while (!rli->relay_log.find_next_log(&linfo, 1));

335

316

/*

336

317

As we have counted everything, including what may have written in a

338

319

twice.

339

320

*/

340

321

rli->relay_log.reset_bytes_written();

341

DBUG_RETURN(0);

322

return(0);

342

323

}

343

324

344

325

352

333

353

334

void Relay_log_info::clear_until_condition()

354

335

{

355

DBUG_ENTER("clear_until_condition");

356

357

336

until_condition= Relay_log_info::UNTIL_NONE;

358

337

until_log_name[0]= 0;

359

338

until_log_pos= 0;

360

DBUG_VOID_RETURN;

339

return;

361

340

}

362

341

363

342

398

377

const char** errmsg,

399

378

bool look_for_description_event)

400

379

{

401

DBUG_ENTER("init_relay_log_pos");

402

DBUG_PRINT("info", ("pos: %lu", (ulong) pos));

403

404

380

*errmsg=0;

405

381

pthread_mutex_t *log_lock=rli->relay_log.get_log_lock();

406

382

491

467

Read the possible Format_description_log_event; if position

492

468

was 4, no need, it will be read naturally.

493

469

*/

494

DBUG_PRINT("info",("looking for a Format_description_log_event"));

495

496

470

if (my_b_tell(rli->cur_log) >= pos)

497

471

break;

498

472

503

477

if (!(ev=Log_event::read_log_event(rli->cur_log,0,

504

478

rli->relay_log.description_event_for_exec)))

505

479

{

506

DBUG_PRINT("info",("could not read event, rli->cur_log->error=%d",

507

rli->cur_log->error));

508

480

if (rli->cur_log->error) /* not EOF */

509

481

{

510

482

*errmsg= "I/O error reading event at position 4";

514

486

}

515

487

else if (ev->get_type_code() == FORMAT_DESCRIPTION_EVENT)

516

488

{

517

DBUG_PRINT("info",("found Format_description_log_event"));

518

489

delete rli->relay_log.description_event_for_exec;

519

490

rli->relay_log.description_event_for_exec= (Format_description_log_event*) ev;

520

491

/*

540

511

}

541

512

else

542

513

{

543

DBUG_PRINT("info",("found event of another type=%d",

544

ev->get_type_code()));

545

514

look_for_description_event= (ev->get_type_code() == ROTATE_EVENT);

546

515

delete ev;

547

516

}

548

517

}

549

518

my_b_seek(rli->cur_log,(off_t)pos);

550

#ifndef DBUG_OFF

551

{

552

char llbuf1[22], llbuf2[22];

553

DBUG_PRINT("info", ("my_b_tell(rli->cur_log)=%s rli->event_relay_log_pos=%s",

554

llstr(my_b_tell(rli->cur_log),llbuf1),

555

llstr(rli->event_relay_log_pos,llbuf2)));

556

}

557

#endif

558

519

559

520

}

560

521

574

535

if (!rli->relay_log.description_event_for_exec->is_valid() && !*errmsg)

575

536

*errmsg= "Invalid Format_description log event; could be out of memory";

576

537

577

DBUG_RETURN ((*errmsg) ? 1 : 0);

538

return ((*errmsg) ? 1 : 0);

578

539

}

579

540

580

541

612

573

int error=0;

613

574

struct timespec abstime; // for timeout checking

614

575

const char *msg;

615

DBUG_ENTER("Relay_log_info::wait_for_pos");

616

576

617

577

if (!inited)

618

DBUG_RETURN(-2);

619

620

DBUG_PRINT("enter",("log_name: '%s' log_pos: %lu timeout: %lu",

621

log_name->c_ptr(), (ulong) log_pos, (ulong) timeout));

578

return(-2);

622

579

623

580

set_timespec(abstime,timeout);

624

581

pthread_mutex_lock(&data_lock);

681

638

bool pos_reached;

682

639

int cmp_result= 0;

683

640

684

DBUG_PRINT("info",

685

("init_abort_pos_wait: %ld abort_pos_wait: %ld",

686

init_abort_pos_wait, abort_pos_wait));

687

DBUG_PRINT("info",("group_master_log_name: '%s' pos: %lu",

688

group_master_log_name, (ulong) group_master_log_pos));

689

690

641

/*

691

642

group_master_log_name can be "", if we are just after a fresh

692

643

replication start or after a CHANGE MASTER TO MASTER_HOST/PORT

732

683

733

684

//wait for master update, with optional timeout.

734

685

735

DBUG_PRINT("info",("Waiting for master update"));

736

686

/*

737

687

We are going to pthread_cond_(timed)wait(); if the SQL thread stops it

738

688

will wake us up.

754

704

}

755

705

else

756

706

pthread_cond_wait(&data_cond, &data_lock);

757

DBUG_PRINT("info",("Got signal of master update or timed out"));

758

707

if (error == ETIMEDOUT || error == ETIME)

759

708

{

760

709

error= -1;

762

711

}

763

712

error=0;

764

713

event_count++;

765

DBUG_PRINT("info",("Testing if killed or SQL thread not running"));

766

714

}

767

715

768

716

err:

769

717

thd->exit_cond(msg);

770

DBUG_PRINT("exit",("killed: %d abort: %d slave_running: %d \

771

improper_arguments: %d timed_out: %d",

772

thd->killed_errno(),

773

(int) (init_abort_pos_wait != abort_pos_wait),

774

(int) slave_running,

775

(int) (error == -2),

776

(int) (error == -1)));

777

718

if (thd->killed || init_abort_pos_wait != abort_pos_wait ||

778

719

!slave_running)

779

720

{

780

721

error= -2;

781

722

}

782

DBUG_RETURN( error ? error : event_count );

723

return( error ? error : event_count );

783

724

}

784

725

785

726

786

727

void Relay_log_info::inc_group_relay_log_pos(ulonglong log_pos,

787

728

bool skip_lock)

788

729

{

789

DBUG_ENTER("Relay_log_info::inc_group_relay_log_pos");

790

791

730

if (!skip_lock)

792

731

pthread_mutex_lock(&data_lock);

793

732

inc_event_relay_log_pos();

827

766

the relay log is not "val".

828

767

With the end_log_pos solution, we avoid computations involving lengthes.

829

768

*/

830

DBUG_PRINT("info", ("log_pos: %lu group_master_log_pos: %lu",

831

(long) log_pos, (long) group_master_log_pos));

832

769

if (log_pos) // 3.23 binlogs don't have log_posx

833

770

{

834

771

group_master_log_pos= log_pos;

836

773

pthread_cond_broadcast(&data_cond);

837

774

if (!skip_lock)

838

775

pthread_mutex_unlock(&data_lock);

839

DBUG_VOID_RETURN;

776

return;

840

777

}

841

778

842

779

843

780

void Relay_log_info::close_temporary_tables()

844

781

{

845

782

TABLE *table,*next;

846

DBUG_ENTER("Relay_log_info::close_temporary_tables");

847

783

848

784

for (table=save_temporary_tables ; table ; table=next)

849

785

{

852

788

Don't ask for disk deletion. For now, anyway they will be deleted when

853

789

slave restarts, but it is a better intention to not delete them.

854

790

*/

855

DBUG_PRINT("info", ("table: 0x%lx", (long) table));

856

791

close_temporary(table, 1, 0);

857

792

}

858

793

save_temporary_tables= 0;

859

794

slave_open_temp_tables= 0;

860

DBUG_VOID_RETURN;

795

return;

861

796

}

862

797

863

798

/*

871

806

const char** errmsg)

872

807

{

873

808

int error=0;

874

DBUG_ENTER("purge_relay_logs");

875

809

876

810

/*

877

811

Even if rli->inited==0, we still try to empty rli->master_log_* variables.

896

830

897

831

if (!rli->inited)

898

832

{

899

DBUG_PRINT("info", ("rli->inited == 0"));

900

DBUG_RETURN(0);

833

return(0);

901

834

}

902

835

903

DBUG_ASSERT(rli->slave_running == 0);

904

DBUG_ASSERT(rli->mi->slave_running == 0);

836

assert(rli->slave_running == 0);

837

assert(rli->mi->slave_running == 0);

905

838

906

839

rli->slave_skip_counter=0;

907

840

pthread_mutex_lock(&rli->data_lock);

942

875

0 /* do not need data lock */, errmsg, 0);

943

876

944

877

err:

945

#ifndef DBUG_OFF

946

char buf[22];

947

#endif

948

DBUG_PRINT("info",("log_space_total: %s",llstr(rli->log_space_total,buf)));

949

878

pthread_mutex_unlock(&rli->data_lock);

950

DBUG_RETURN(error);

879

return(error);

951

880

}

952

881

953

882

988

917

{

989

918

const char *log_name;

990

919

ulonglong log_pos;

991

DBUG_ENTER("Relay_log_info::is_until_satisfied");

992

920

993

DBUG_ASSERT(until_condition != UNTIL_NONE);

921

assert(until_condition != UNTIL_NONE);

994

922

995

923

if (until_condition == UNTIL_MASTER_POS)

996

924

{

1003

931

log_pos= group_relay_log_pos;

1004

932

}

1005

933

1006

#ifndef DBUG_OFF

1007

{

1008

char buf[32];

1009

DBUG_PRINT("info", ("group_master_log_name='%s', group_master_log_pos=%s",

1010

group_master_log_name, llstr(group_master_log_pos, buf)));

1011

DBUG_PRINT("info", ("group_relay_log_name='%s', group_relay_log_pos=%s",

1012

group_relay_log_name, llstr(group_relay_log_pos, buf)));

1013

DBUG_PRINT("info", ("(%s) log_name='%s', log_pos=%s",

1014

until_condition == UNTIL_MASTER_POS ? "master" : "relay",

1015

log_name, llstr(log_pos, buf)));

1016

DBUG_PRINT("info", ("(%s) until_log_name='%s', until_log_pos=%s",

1017

until_condition == UNTIL_MASTER_POS ? "master" : "relay",

1018

until_log_name, llstr(until_log_pos, buf)));

1019

}

1020

#endif

1021

1022

934

if (until_log_names_cmp_result == UNTIL_LOG_NAMES_CMP_UNKNOWN)

1023

935

{

1024

936

/*

1051

963

/* Probably error so we aborting */

1052

964

sql_print_error("Slave SQL thread is stopped because UNTIL "

1053

965

"condition is bad.");

1054

DBUG_RETURN(TRUE);

966

return(true);

1055

967

}

1056

968

}

1057

969

else

1058

DBUG_RETURN(until_log_pos == 0);

970

return(until_log_pos == 0);

1059

971

}

1060

972

1061

DBUG_RETURN(((until_log_names_cmp_result == UNTIL_LOG_NAMES_CMP_EQUAL &&

973

return(((until_log_names_cmp_result == UNTIL_LOG_NAMES_CMP_EQUAL &&

1062

974

log_pos >= until_log_pos) ||

1063

975

until_log_names_cmp_result == UNTIL_LOG_NAMES_CMP_GREATER));

1064

976

}

1066

978

1067

979

void Relay_log_info::cached_charset_invalidate()

1068

980

{

1069

DBUG_ENTER("Relay_log_info::cached_charset_invalidate");

1070

1071

981

/* Full of zeroes means uninitialized. */

1072

982

bzero(cached_charset, sizeof(cached_charset));

1073

DBUG_VOID_RETURN;

983

return;

1074

984

}

1075

985

1076

986

1077

987

bool Relay_log_info::cached_charset_compare(char *charset) const

1078

988

{

1079

DBUG_ENTER("Relay_log_info::cached_charset_compare");

1080

1081

989

if (bcmp((uchar*) cached_charset, (uchar*) charset,

1082

990

sizeof(cached_charset)))

1083

991

{

1084

992

memcpy(const_cast<char*>(cached_charset), charset, sizeof(cached_charset));

1085

DBUG_RETURN(1);

993

return(1);

1086

994

}

1087

DBUG_RETURN(0);

995

return(0);

1088

996

}

1089

997

1090

998

1091

999

void Relay_log_info::stmt_done(my_off_t event_master_log_pos,

1092

1000

time_t event_creation_time)

1093

1001

{

1094

#ifndef DBUG_OFF

1095

1002

extern uint debug_not_change_ts_if_art_event;

1096

#endif

1097

1003

clear_flag(IN_STMT);

1098

1004

1099

1005

/*

1133

1039

is that value may take some time to display in

1134

1040

Seconds_Behind_Master - not critical).

1135

1041

*/

1136

#ifndef DBUG_OFF

1137

1042

if (!(event_creation_time == 0 && debug_not_change_ts_if_art_event > 0))

1138

#else

1139

if (event_creation_time != 0)

1140

#endif

1141

last_master_timestamp= event_creation_time;

1043

last_master_timestamp= event_creation_time;

1142

1044

}

1143

1045

}

1144

1046

1145

1047

#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)

1146

1048

void Relay_log_info::cleanup_context(THD *thd, bool error)

1147

1049

{

1148

DBUG_ENTER("Relay_log_info::cleanup_context");

1149

1150

DBUG_ASSERT(sql_thd == thd);

1050

assert(sql_thd == thd);

1151

1051

/*

1152

1052

1) Instances of Table_map_log_event, if ::do_apply_event() was called on them,

1153

1053

may have opened tables, which we cannot be sure have been closed (because

1175

1075

thd->options&= ~OPTION_NO_FOREIGN_KEY_CHECKS;

1176

1076

thd->options&= ~OPTION_RELAXED_UNIQUE_CHECKS;

1177

1077

last_event_start_time= 0;

1178

DBUG_VOID_RETURN;

1078

return;

1179

1079

}

1180

1080

1181

1081

void Relay_log_info::clear_tables_to_lock()

1186

1086

if (tables_to_lock->m_tabledef_valid)

1187

1087

{

1188

1088

tables_to_lock->m_tabledef.table_def::~table_def();

1189

tables_to_lock->m_tabledef_valid= FALSE;

1089

tables_to_lock->m_tabledef_valid= false;

1190

1090

}

1191

1091

tables_to_lock=

1192

1092

static_cast<RPL_TABLE_LIST*>(tables_to_lock->next_global);

1193

1093

tables_to_lock_count--;

1194

1094

my_free(to_free, MYF(MY_WME));

1195

1095

}

1196

DBUG_ASSERT(tables_to_lock == NULL && tables_to_lock_count == 0);

1096

assert(tables_to_lock == NULL && tables_to_lock_count == 0);

1197

1097

}

1198

1098

1199

1099

#endif