~drizzle-trunk/drizzle/development

if ((scanned != 8) || (blob->bu_type != MS_URL_TYPE_BLOB && blob->bu_type != MS_URL_TYPE_REPO)) {// If junk is found at the end this will also result in an invalid URL.

192

//printf("Bad URL \"%s\": scanned = %d, junk: %d, %d, %d, %d\n", blob_url, scanned, junk[0], junk[1], junk[2], junk[3]);

193

return false;

194

}

195

196

return true;

197

}

198

199

return false;

200

}

201

202

static bool couldBeURL(const char *blob_url, MSBlobURLPtr blob)

203

{

204

return couldBeURL(blob_url, strlen(blob_url), blob);

205

}

206

207

static void buildBlobURL(MSBlobURLPtr blob, PBMSBlobURLPtr url)

208

{

209

snprintf(url->bu_data, PBMS_BLOB_URL_SIZE, URL_FMT, blob->bu_db_id,

210

blob->bu_type,

211

blob->bu_tab_id,

212

blob->bu_blob_id,

213

blob->bu_auth_code,

214

blob->bu_server_id,

215

blob->bu_blob_ref_id,

216

blob->bu_blob_size);

217

}

218

};

219

220

#ifndef DRIZZLED

221

222

* This function should never be called directly, it is called

223

* by deregisterEngine() below.

224

225

typedef void (*ECRegisterdFunc)(PBMSEnginePtr engine);

226

227

typedef void (*ECDeregisterdFunc)(PBMSEnginePtr engine);

228

229

230

* Call this function to store a BLOB in the repository the BLOB's

231

* URL will be returned. The returned URL buffer is expected to be atleast

232

* PBMS_BLOB_URL_SIZE long.

233

234

* The BLOB URL must still be retained or it will automaticly be deleted after a timeout expires.

235

236

typedef int (*ECCreateBlobsFunc)(bool built_in, const char *db_name, const char *tab_name, char *blob, size_t blob_len, PBMSBlobURLPtr blob_url, PBMSResultPtr result);

237

238

239

* Call this function for each BLOB to be retained. When a BLOB is used, the

240

* URL may be changed. The returned URL buffer is expected to be atleast

241

* PBMS_BLOB_URL_SIZE long.

242

243

* The returned URL must be inserted into the row in place of the given

244

* URL.

245

246

typedef int (*ECRetainBlobsFunc)(bool built_in, const char *db_name, const char *tab_name, PBMSBlobURLPtr ret_blob_url, char *blob_url, unsigned short col_index, PBMSResultPtr result);

247

248

249

* If a row containing a BLOB is deleted, then the BLOBs in the

250

* row must be released.

251

252

* Note: if a table is dropped, all the BLOBs referenced by the

253

* table are automatically released.

254

255

typedef int (*ECReleaseBlobFunc)(bool built_in, const char *db_name, const char *tab_name, char *blob_url, PBMSResultPtr result);

256

257

typedef int (*ECDropTable)(bool built_in, const char *db_name, const char *tab_name, PBMSResultPtr result);

258

259

typedef int (*ECRenameTable)(bool built_in, const char *db_name, const char *from_table, const char *to_db, const char *to_table, PBMSResultPtr result);

260

261

typedef void (*ECCallCompleted)(bool built_in, bool ok);

262

263

typedef struct PBMSCallbacksRec {

264

int cb_version; /* MS_CALLBACK_VERSION */

265

ECRegisterdFunc cb_register;

266

ECDeregisterdFunc cb_deregister;

267

ECCreateBlobsFunc cb_create_blob;

268

ECRetainBlobsFunc cb_retain_blob;

269

ECReleaseBlobFunc cb_release_blob;

270

ECDropTable cb_drop_table;

271

ECRenameTable cb_rename_table;

272

ECCallCompleted cb_completed;

273

} PBMSCallbacksRec, *PBMSCallbacksPtr;

274

275

typedef struct PBMSSharedMemoryRec {

276

int sm_magic; /* MS_SHARED_MEMORY_MAGIC */

277

int sm_version; /* MS_SHARED_MEMORY_VERSION */

278

volatile int sm_shutdown_lock; /* "Cheap" lock for shutdown! */

279

PBMSCallbacksPtr sm_callbacks;

280

int sm_reserved1[20];

281

void *sm_reserved2[20];

282

int sm_list_size;

283

int sm_list_len;

284

PBMSEnginePtr sm_engine_list[MS_ENGINE_LIST_SIZE];

285

} PBMSSharedMemoryRec, *PBMSSharedMemoryPtr;

286

287

#ifdef PBMS_API

288

289

class PBMS_API

290

{

291

private:

292

const char *temp_prefix[3];

293

bool built_in;

294

295

public:

296

PBMS_API(): sharedMemory(NULL) {

297

int i = 0;

298

temp_prefix[i++] = MS_TEMP_FILE_PREFIX;

299

temp_prefix[i++] = NULL;

300

301

}

302

303

~PBMS_API() { }

304

305

306

* This method is called by the PBMS engine during startup.

307

308

int PBMSStartup(PBMSCallbacksPtr callbacks, PBMSResultPtr result) {

309

int err;

310

311

deleteTempFiles();

312

err = getSharedMemory(true, result);

313

if (!err)

314

sharedMemory->sm_callbacks = callbacks;

315

316

return err;

317

}

318

319

320

* This method is called by the PBMS engine during startup.

321

322

void PBMSShutdown() {

323

324

if (!sharedMemory)

325

return;

326

327

lock();

328

sharedMemory->sm_callbacks = NULL;

329

330

bool empty = true;

331

for (int i=0; i<sharedMemory->sm_list_len && empty; i++) {

332

if (sharedMemory->sm_engine_list[i])

333

empty = false;

334

}

335

336

unlock();

337

338

if (empty)

339

removeSharedMemory();

340

}

341

342

343

* Register the engine with the Stream Daemon.

344

345

int registerEngine(PBMSEnginePtr the_engine, PBMSResultPtr result) {

346

int err;

347

348

deleteTempFiles();

349

350

// The first engine to register creates the shared memory.

351

if ((err = getSharedMemory(true, result)))

352

return err;

353

354

lock();

355

for (int i=0; i<sharedMemory->sm_list_size; i++) {

356

if (!sharedMemory->sm_engine_list[i]) {

357

PBMSEnginePtr engine;

358

engine = (PBMSEnginePtr) malloc(sizeof(PBMSEngineRec));

359

if (!engine) {

360

strcpy(MS_RESULT_MESSAGE_SIZE, result->mr_message, "Out of memory.");

361

err = MS_ERR_ENGINE;

362

goto done;

363

}

364

memcpy(engine, the_engine, sizeof(PBMSEngineRec));

365

366

sharedMemory->sm_engine_list[i] = engine;

367

engine->ms_index = i;

368

if (i >= sharedMemory->sm_list_len)

369

sharedMemory->sm_list_len = i+1;

370

if (sharedMemory->sm_callbacks)

371

sharedMemory->sm_callbacks->cb_register(engine);

372

373

built_in = (engine->ms_internal == 1);

374

err = MS_OK;

375

goto done;

376

}

377

}

378

379

result->mr_code = 15010;

380

strcpy(MS_RESULT_MESSAGE_SIZE, result->mr_message, "Too many BLOB streaming engines already registered");

381

*result->mr_stack = 0;

382

383

err = MS_ERR_ENGINE;

384

385

done:

386

unlock();

387

return err;

388

}

389

390

void lock() {

391

while (sharedMemory->sm_shutdown_lock)

392

usleep(10000);

393

sharedMemory->sm_shutdown_lock++;

394

while (sharedMemory->sm_shutdown_lock != 1) {

395

usleep(random() % 10000);

396

sharedMemory->sm_shutdown_lock--;

397

usleep(10000);

398

sharedMemory->sm_shutdown_lock++;

399

}

400

}

401

402

void unlock() {

403

sharedMemory->sm_shutdown_lock--;

404

}

405

406

void deregisterEngine(const char *engine_name) {

407

PBMSResultRec result;

408

int err;

409

410

if ((err = getSharedMemory(false, &result)))

411

return;

412

413

lock();

414

415

bool empty = true;

416

for (int i=0; i<sharedMemory->sm_list_len; i++) {

417

PBMSEnginePtr engine = sharedMemory->sm_engine_list[i];

418

if (engine) {

419

if (strcmp(engine->ms_engine_name, engine_name) == 0) {

420

if (sharedMemory->sm_callbacks)

421

sharedMemory->sm_callbacks->cb_deregister(engine);

422

free(engine);

423

sharedMemory->sm_engine_list[i] = NULL;

424

}

425

else

426

empty = false;

427

}

428

}

429

430

unlock();

431

432

if (empty)

433

removeSharedMemory();

434

}

435

436

void removeSharedMemory()

437

{

438

const char **prefix = temp_prefix;

439

char temp_file[100];

440

441

// Do not remove the sharfed memory until after

442

// the PBMS engine has shutdown.

443

if (sharedMemory->sm_callbacks)

444

return;

445

446

sharedMemory->sm_magic = 0;

447

free(sharedMemory);

448

sharedMemory = NULL;

449

450

while (*prefix) {

451

getTempFileName(temp_file, 100, *prefix, getpid());

452

unlink(temp_file);

453

prefix++;

454

}

455

}

456

457

bool isPBMSLoaded()

458

{

459

PBMSResultRec result;

460

if (getSharedMemory(false, &result))

461

return false;

462

463

return (sharedMemory->sm_callbacks != NULL);

464

}

465

466

int retainBlob(const char *db_name, const char *tab_name, PBMSBlobURLPtr ret_blob_url, char *blob_url, size_t blob_size, unsigned short col_index, PBMSResultPtr result)

467

{

468

int err;

469

char safe_url[PBMS_BLOB_URL_SIZE+1];

470

471

472

if ((err = getSharedMemory(false, result)))

473

return err;

474

475

if (!PBMSBlobURLTools::couldBeURL(blob_url, blob_size, NULL)) {

476

477

if (!sharedMemory->sm_callbacks) {

478

ret_blob_url->bu_data[0] = 0;

479

return MS_OK;

480

}

481

err = sharedMemory->sm_callbacks->cb_create_blob(built_in, db_name, tab_name, blob_url, blob_size, ret_blob_url, result);

482

if (err)

483

return err;

484

485

blob_url = ret_blob_url->bu_data;

486

} else {

487

// Make sure the url is a C string:

488

if (blob_url[blob_size]) {

489

memcpy(safe_url, blob_url, blob_size);

490

safe_url[blob_size] = 0;

491

blob_url = safe_url;

492

}

493

}

494

495

496

if (!sharedMemory->sm_callbacks) {

497

result->mr_code = MS_ERR_INCORRECT_URL;

498

strcpy(MS_RESULT_MESSAGE_SIZE, result->mr_message, "BLOB streaming daemon (PBMS) not installed");

499

*result->mr_stack = 0;

500

return MS_ERR_INCORRECT_URL;

501

}

502

503

return sharedMemory->sm_callbacks->cb_retain_blob(built_in, db_name, tab_name, ret_blob_url, blob_url, col_index, result);

504

}

505

506

int releaseBlob(const char *db_name, const char *tab_name, char *blob_url, size_t blob_size, PBMSResultPtr result)

507

{

508

int err;

509

char safe_url[PBMS_BLOB_URL_SIZE+1];

510

511

if ((err = getSharedMemory(false, result)))

512

return err;

513

514

if (!sharedMemory->sm_callbacks)

515

return MS_OK;

516

517

if (!PBMSBlobURLTools::couldBeURL(blob_url, blob_size, NULL))

518

return MS_OK;

519

520

if (blob_url[blob_size]) {

521

memcpy(safe_url, blob_url, blob_size);

522

safe_url[blob_size] = 0;

523

blob_url = safe_url;

524

}

525

526

return sharedMemory->sm_callbacks->cb_release_blob(built_in, db_name, tab_name, blob_url, result);

527

}

528

529

int dropTable(const char *db_name, const char *tab_name, PBMSResultPtr result)

530

{

531

int err;

532

533

if ((err = getSharedMemory(false, result)))

534

return err;

535

536

if (!sharedMemory->sm_callbacks)

537

return MS_OK;

538

539

return sharedMemory->sm_callbacks->cb_drop_table(built_in, db_name, tab_name, result);

540

}

541

542

int renameTable(const char *db_name, const char *from_table, const char *to_db, const char *to_table, PBMSResultPtr result)

543

{

544

int err;

545

546

if ((err = getSharedMemory(false, result)))

547

return err;

548

549

if (!sharedMemory->sm_callbacks)

550

return MS_OK;

551

552

return sharedMemory->sm_callbacks->cb_rename_table(built_in, db_name, from_table, to_db, to_table, result);

553

}

554

555

void completed(int ok)

556

{

557

PBMSResultRec result;

558

559

if (getSharedMemory(false, &result))

560

return;

561

562

if (!sharedMemory->sm_callbacks)

563

return;

564

565

sharedMemory->sm_callbacks->cb_completed(built_in, ok);

566

}

567

568

volatile PBMSSharedMemoryPtr sharedMemory;

569

570

private:

571

int getSharedMemory(bool create, PBMSResultPtr result)

572

{

573

int tmp_f;

574

int r;

575

char temp_file[100];

576

const char **prefix = temp_prefix;

577

578

if (sharedMemory)

579

return MS_OK;

580

581

while (*prefix) {

582

getTempFileName(temp_file, 100, *prefix, getpid());

583

tmp_f = open(temp_file, O_RDWR | (create ? O_CREAT : 0), SH_MASK);

584

if (tmp_f == -1)

585

return setOSResult(errno, "open", temp_file, result);

586

587

r = lseek(tmp_f, 0, SEEK_SET);

588

if (r == -1) {

589

close(tmp_f);

590

return setOSResult(errno, "lseek", temp_file, result);

591

}

592

ssize_t tfer;

593

char buffer[100];

594

595

tfer = read(tmp_f, buffer, 100);

596

if (tfer == -1) {

597

close(tmp_f);

598

return setOSResult(errno, "read", temp_file, result);

599

}

600

601

buffer[tfer] = 0;

602

sscanf(buffer, "%p", (void**) &sharedMemory);

603

if (!sharedMemory || sharedMemory->sm_magic != MS_SHARED_MEMORY_MAGIC) {

604

if (!create)

605

return MS_OK;

606

607

sharedMemory = (PBMSSharedMemoryPtr) calloc(1, sizeof(PBMSSharedMemoryRec));

608

sharedMemory->sm_magic = MS_SHARED_MEMORY_MAGIC;

609

sharedMemory->sm_version = MS_SHARED_MEMORY_VERSION;

610

sharedMemory->sm_list_size = MS_ENGINE_LIST_SIZE;

611

612

r = lseek(tmp_f, 0, SEEK_SET);

613

if (r == -1) {

614

close(tmp_f);

615

return setOSResult(errno, "fseek", temp_file, result);

616

}

617

618

snprintf(buffer, 100, "%p", (void*) sharedMemory);

619

tfer = write(tmp_f, buffer, strlen(buffer));

620

if (tfer != (ssize_t) strlen(buffer)) {

621

close(tmp_f);

622

return setOSResult(errno, "write", temp_file, result);

623

}

624

r = fsync(tmp_f);

625

if (r == -1) {

626

close(tmp_f);

627

return setOSResult(errno, "fsync", temp_file, result);

628

}

629

}

630

else if (sharedMemory->sm_version != MS_SHARED_MEMORY_VERSION) {

631

close(tmp_f);

632

result->mr_code = -1000;

633

*result->mr_stack = 0;

634

strcpy(MS_RESULT_MESSAGE_SIZE, result->mr_message, "Shared memory version: ");

635

strcat(MS_RESULT_MESSAGE_SIZE, result->mr_message, sharedMemory->sm_version);

636

strcat(MS_RESULT_MESSAGE_SIZE, result->mr_message, ", does not match engine shared memory version: ");

637

strcat(MS_RESULT_MESSAGE_SIZE, result->mr_message, MS_SHARED_MEMORY_VERSION);

638

strcat(MS_RESULT_MESSAGE_SIZE, result->mr_message, ".");

639

return MS_ERR_ENGINE;

640

}

641

close(tmp_f);

642

643

// For backward compatability we need to create the old versions but we only need to read the current version.

644

if (create)

645

prefix++;

646

else

647

break;

648

}

649

return MS_OK;

650

}

651

652

void strcpy(size_t size, char *to, const char *from)

653

{

654

if (size > 0) {

655

size--;

656

while (*from && size--)

657

*to++ = *from++;

658

*to = 0;

659

}

660

}

661

662

void strcat(size_t size, char *to, const char *from)

663

{

664

while (*to && size--) to++;

665

strcpy(size, to, from);

666

}

667

668

void strcat(size_t size, char *to, int val)

669

{

670

char buffer[20];

671

672

snprintf(buffer, 20, "%d", val);

673

strcat(size, to, buffer);

674

}

675

676

int setOSResult(int err, const char *func, char *file, PBMSResultPtr result) {

677

char *msg;

678

679

result->mr_code = err;

680

*result->mr_stack = 0;

681

strcpy(MS_RESULT_MESSAGE_SIZE, result->mr_message, "System call ");

682

strcat(MS_RESULT_MESSAGE_SIZE, result->mr_message, func);

683

strcat(MS_RESULT_MESSAGE_SIZE, result->mr_message, "() failed on ");

684

strcat(MS_RESULT_MESSAGE_SIZE, result->mr_message, file);

685

strcat(MS_RESULT_MESSAGE_SIZE, result->mr_message, ": ");

686

687

#ifdef XT_WIN

688

if (FormatMessage(FORMAT_MESSAGE_FROM_SYSTEM, NULL, err, 0, iMessage + strlen(iMessage), MS_RESULT_MESSAGE_SIZE - strlen(iMessage), NULL)) {

689

char *ptr;

690

691

ptr = &iMessage[strlen(iMessage)];

692

while (ptr-1 > err_msg) {

693

if (*(ptr-1) != '\n' && *(ptr-1) != '\r' && *(ptr-1) != '.')

694

break;

695

ptr--;

696

}

697

*ptr = 0;

698

699

strcat(MS_RESULT_MESSAGE_SIZE, result->mr_message, " (");

700

strcat(MS_RESULT_MESSAGE_SIZE, result->mr_message, err);

701

strcat(MS_RESULT_MESSAGE_SIZE, result->mr_message, ")");

702

return MS_ERR_ENGINE;

703

}

704

#endif

705

706

msg = strerror(err);

707

if (msg) {

708

strcat(MS_RESULT_MESSAGE_SIZE, result->mr_message, msg);

709

strcat(MS_RESULT_MESSAGE_SIZE, result->mr_message, " (");

710

strcat(MS_RESULT_MESSAGE_SIZE, result->mr_message, err);

711

strcat(MS_RESULT_MESSAGE_SIZE, result->mr_message, ")");

712

}

713

else {

714

strcat(MS_RESULT_MESSAGE_SIZE, result->mr_message, "Unknown OS error code ");

715

strcat(MS_RESULT_MESSAGE_SIZE, result->mr_message, err);

716

}

717

718

return MS_ERR_ENGINE;

719

}

720

721

void getTempFileName(char *temp_file, int buffer_size, const char * prefix, int pid)

722

{

723

snprintf(temp_file, buffer_size, "/tmp/%s%d", prefix, pid);

724

}

725

726

bool startsWith(const char *cstr, const char *w_cstr)

727

{

728

while (*cstr && *w_cstr) {

729

if (*cstr != *w_cstr)

730

return false;

731

cstr++;

732

w_cstr++;

733

}

734

return *cstr || !*w_cstr;

735

}

736

737

void deleteTempFiles()

738

{

739

struct dirent entry;

740

struct dirent *result;

741

DIR *odir;

742

int err;

743

char temp_file[100];

744

745

if (!(odir = opendir("/tmp/")))

746

return;

747

err = readdir_r(odir, &entry, &result);

748

while (!err && result) {

749

const char **prefix = temp_prefix;

750

751

while (*prefix) {

752

if (startsWith(entry.d_name, *prefix)) {

753

int pid = atoi(entry.d_name + strlen(*prefix));

754

755

/* If the process does not exist: */

756

if (kill(pid, 0) == -1 && errno == ESRCH) {

757

getTempFileName(temp_file, 100, *prefix, pid);

758

unlink(temp_file);

759

}

760

}

761

prefix++;

762

}

763

764

err = readdir_r(odir, &entry, &result);

765

}

766

closedir(odir);

767

}

768

};

769

#endif // PBMS_API

770

771

772

* The following is a low level API for accessing blobs directly.

773

774

775

776

777

* Any threads using the direct blob access API must first register them selves with the

778

* blob streaming engine before using the blob access functions. This is done by calling

779

* PBMSInitBlobStreamingThread(). Call PBMSDeinitBlobStreamingThread() after the thread is

780

* done using the direct blob access API

781

782

783

784

* PBMSInitBlobStreamingThread(): Returns a pointer to a blob streaming thread.

785

786

extern void *PBMSInitBlobStreamingThread(char *thread_name, PBMSResultPtr result);

787

extern void PBMSDeinitBlobStreamingThread(void *v_bs_thread);

788

789

790

* PBMSGetError():Gets the last error reported by a blob streaming thread.

791

792

extern void PBMSGetError(void *v_bs_thread, PBMSResultPtr result);

793

794

795

* PBMSCreateBlob():Creates a new blob in the database of the given size.

796

797

extern bool PBMSCreateBlob(PBMSBlobIDPtr blob_id, char *database_name, u_int64_t size);

798

799

800

* PBMSWriteBlob():Write the data to the blob in one or more chunks. The total size of all the chuncks of

801

* data written to the blob must match the size specified when the blob was created.

802

803

extern bool PBMSWriteBlob(PBMSBlobIDPtr blob_id, char *data, size_t size, size_t offset);

804

805

806

* PBMSReadBlob():Read the blob data out of the blob in one or more chunks.

807

808

extern bool PBMSReadBlob(PBMSBlobIDPtr blob_id, char *buffer, size_t *size, size_t offset);

809

810

811

* PBMSIDToURL():Convert a blob id to a blob URL. The 'url' buffer must be atleast PBMS_BLOB_URL_SIZE bytes in size.

812

813

extern bool PBMSIDToURL(PBMSBlobIDPtr blob_id, char *url);

814

815

816

* PBMSIDToURL():Convert a blob URL to a blob ID.

817

818

extern bool PBMSURLToID(char *url, PBMSBlobIDPtr blob_id);

819

#endif //DRIZZLED

820

821

822

#endif //__PBMS_H__

Older »