~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.

194

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

195

return false;

196

}

197

198

return true;

199

}

200

201

return false;

202

}

203

204

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

205

{

206

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

207

}

208

209

static void buildBlobURL(MSBlobURLPtr blob, PBMSBlobURLPtr url)

210

{

211

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

212

blob->bu_type,

213

blob->bu_tab_id,

214

blob->bu_blob_id,

215

blob->bu_auth_code,

216

blob->bu_server_id,

217

blob->bu_blob_ref_id,

218

blob->bu_blob_size);

219

}

220

};

221

222

#ifndef DRIZZLED

223

224

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

225

* by deregisterEngine() below.

226

227

typedef void (*ECRegisterdFunc)(PBMSEnginePtr engine);

228

229

typedef void (*ECDeregisterdFunc)(PBMSEnginePtr engine);

230

231

232

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

233

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

234

* PBMS_BLOB_URL_SIZE long.

235

236

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

237

238

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);

239

240

241

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

242

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

243

* PBMS_BLOB_URL_SIZE long.

244

245

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

246

* URL.

247

248

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);

249

250

251

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

252

* row must be released.

253

254

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

255

* table are automatically released.

256

257

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

258

259

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

260

261

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

262

263

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

264

265

typedef struct PBMSCallbacksRec {

266

int cb_version; /* MS_CALLBACK_VERSION */

267

ECRegisterdFunc cb_register;

268

ECDeregisterdFunc cb_deregister;

269

ECCreateBlobsFunc cb_create_blob;

270

ECRetainBlobsFunc cb_retain_blob;

271

ECReleaseBlobFunc cb_release_blob;

272

ECDropTable cb_drop_table;

273

ECRenameTable cb_rename_table;

274

ECCallCompleted cb_completed;

275

} PBMSCallbacksRec, *PBMSCallbacksPtr;

276

277

typedef struct PBMSSharedMemoryRec {

278

int sm_magic; /* MS_SHARED_MEMORY_MAGIC */

279

int sm_version; /* MS_SHARED_MEMORY_VERSION */

280

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

281

PBMSCallbacksPtr sm_callbacks;

282

int sm_reserved1[20];

283

void *sm_reserved2[20];

284

int sm_list_size;

285

int sm_list_len;

286

PBMSEnginePtr sm_engine_list[MS_ENGINE_LIST_SIZE];

287

} PBMSSharedMemoryRec, *PBMSSharedMemoryPtr;

288

289

#ifdef PBMS_API

290

291

class PBMS_API

292

{

293

private:

294

const char *temp_prefix[3];

295

bool built_in;

296

297

public:

298

PBMS_API(): sharedMemory(NULL) {

299

int i = 0;

300

temp_prefix[i++] = MS_TEMP_FILE_PREFIX;

301

temp_prefix[i++] = NULL;

302

303

}

304

305

~PBMS_API() { }

306

307

308

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

309

310

int PBMSStartup(PBMSCallbacksPtr callbacks, PBMSResultPtr result) {

311

int err;

312

313

deleteTempFiles();

314

err = getSharedMemory(true, result);

315

if (!err)

316

sharedMemory->sm_callbacks = callbacks;

317

318

return err;

319

}

320

321

322

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

323

324

void PBMSShutdown() {

325

326

if (!sharedMemory)

327

return;

328

329

lock();

330

sharedMemory->sm_callbacks = NULL;

331

332

bool empty = true;

333

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

334

if (sharedMemory->sm_engine_list[i])

335

empty = false;

336

}

337

338

unlock();

339

340

if (empty)

341

removeSharedMemory();

342

}

343

344

345

* Register the engine with the Stream Daemon.

346

347

int registerEngine(PBMSEnginePtr the_engine, PBMSResultPtr result) {

348

int err;

349

350

deleteTempFiles();

351

352

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

353

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

354

return err;

355

356

lock();

357

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

358

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

359

PBMSEnginePtr engine;

360

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

361

if (!engine) {

362

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

363

err = MS_ERR_ENGINE;

364

goto done;

365

}

366

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

367

368

sharedMemory->sm_engine_list[i] = engine;

369

engine->ms_index = i;

370

if (i >= sharedMemory->sm_list_len)

371

sharedMemory->sm_list_len = i+1;

372

if (sharedMemory->sm_callbacks)

373

sharedMemory->sm_callbacks->cb_register(engine);

374

375

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

376

err = MS_OK;

377

goto done;

378

}

379

}

380

381

result->mr_code = 15010;

382

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

383

*result->mr_stack = 0;

384

385

err = MS_ERR_ENGINE;

386

387

done:

388

unlock();

389

return err;

390

}

391

392

void lock() {

393

while (sharedMemory->sm_shutdown_lock)

394

usleep(10000);

395

sharedMemory->sm_shutdown_lock++;

396

while (sharedMemory->sm_shutdown_lock != 1) {

397

usleep(random() % 10000);

398

sharedMemory->sm_shutdown_lock--;

399

usleep(10000);

400

sharedMemory->sm_shutdown_lock++;

401

}

402

}

403

404

void unlock() {

405

sharedMemory->sm_shutdown_lock--;

406

}

407

408

void deregisterEngine(const char *engine_name) {

409

PBMSResultRec result;

410

int err;

411

412

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

413

return;

414

415

lock();

416

417

bool empty = true;

418

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

419

PBMSEnginePtr engine = sharedMemory->sm_engine_list[i];

420

if (engine) {

421

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

422

if (sharedMemory->sm_callbacks)

423

sharedMemory->sm_callbacks->cb_deregister(engine);

424

free(engine);

425

sharedMemory->sm_engine_list[i] = NULL;

426

}

427

else

428

empty = false;

429

}

430

}

431

432

unlock();

433

434

if (empty)

435

removeSharedMemory();

436

}

437

438

void removeSharedMemory()

439

{

440

const char **prefix = temp_prefix;

441

char temp_file[100];

442

443

// Do not remove the sharfed memory until after

444

// the PBMS engine has shutdown.

445

if (sharedMemory->sm_callbacks)

446

return;

447

448

sharedMemory->sm_magic = 0;

449

free(sharedMemory);

450

sharedMemory = NULL;

451

452

while (*prefix) {

453

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

454

unlink(temp_file);

455

prefix++;

456

}

457

}

458

459

bool isPBMSLoaded()

460

{

461

PBMSResultRec result;

462

if (getSharedMemory(false, &result))

463

return false;

464

465

return (sharedMemory->sm_callbacks != NULL);

466

}

467

468

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)

469

{

470

int err;

471

char safe_url[PBMS_BLOB_URL_SIZE+1];

472

473

474

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

475

return err;

476

477

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

478

479

if (!sharedMemory->sm_callbacks) {

480

ret_blob_url->bu_data[0] = 0;

481

return MS_OK;

482

}

483

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

484

if (err)

485

return err;

486

487

blob_url = ret_blob_url->bu_data;

488

} else {

489

// Make sure the url is a C string:

490

if (blob_url[blob_size]) {

491

memcpy(safe_url, blob_url, blob_size);

492

safe_url[blob_size] = 0;

493

blob_url = safe_url;

494

}

495

}

496

497

498

if (!sharedMemory->sm_callbacks) {

499

result->mr_code = MS_ERR_INCORRECT_URL;

500

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

501

*result->mr_stack = 0;

502

return MS_ERR_INCORRECT_URL;

503

}

504

505

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

506

}

507

508

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

509

{

510

int err;

511

char safe_url[PBMS_BLOB_URL_SIZE+1];

512

513

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

514

return err;

515

516

if (!sharedMemory->sm_callbacks)

517

return MS_OK;

518

519

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

520

return MS_OK;

521

522

if (blob_url[blob_size]) {

523

memcpy(safe_url, blob_url, blob_size);

524

safe_url[blob_size] = 0;

525

blob_url = safe_url;

526

}

527

528

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

529

}

530

531

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

532

{

533

int err;

534

535

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

536

return err;

537

538

if (!sharedMemory->sm_callbacks)

539

return MS_OK;

540

541

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

542

}

543

544

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

545

{

546

int err;

547

548

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

549

return err;

550

551

if (!sharedMemory->sm_callbacks)

552

return MS_OK;

553

554

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

555

}

556

557

void completed(int ok)

558

{

559

PBMSResultRec result;

560

561

if (getSharedMemory(false, &result))

562

return;

563

564

if (!sharedMemory->sm_callbacks)

565

return;

566

567

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

568

}

569

570

volatile PBMSSharedMemoryPtr sharedMemory;

571

572

private:

573

int getSharedMemory(bool create, PBMSResultPtr result)

574

{

575

int tmp_f;

576

int r;

577

char temp_file[100];

578

const char **prefix = temp_prefix;

579

580

if (sharedMemory)

581

return MS_OK;

582

583

while (*prefix) {

584

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

585

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

586

if (tmp_f == -1)

587

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

588

589

r = lseek(tmp_f, 0, SEEK_SET);

590

if (r == -1) {

591

close(tmp_f);

592

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

593

}

594

ssize_t tfer;

595

char buffer[100];

596

597

tfer = read(tmp_f, buffer, 100);

598

if (tfer == -1) {

599

close(tmp_f);

600

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

601

}

602

603

buffer[tfer] = 0;

604

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

605

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

606

if (!create)

607

return MS_OK;

608

609

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

610

sharedMemory->sm_magic = MS_SHARED_MEMORY_MAGIC;

611

sharedMemory->sm_version = MS_SHARED_MEMORY_VERSION;

612

sharedMemory->sm_list_size = MS_ENGINE_LIST_SIZE;

613

614

r = lseek(tmp_f, 0, SEEK_SET);

615

if (r == -1) {

616

close(tmp_f);

617

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

618

}

619

620

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

621

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

622

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

623

close(tmp_f);

624

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

625

}

626

r = fsync(tmp_f);

627

if (r == -1) {

628

close(tmp_f);

629

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

630

}

631

}

632

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

633

close(tmp_f);

634

result->mr_code = -1000;

635

*result->mr_stack = 0;

636

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

637

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

638

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

639

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

640

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

641

return MS_ERR_ENGINE;

642

}

643

close(tmp_f);

644

645

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

646

if (create)

647

prefix++;

648

else

649

break;

650

}

651

return MS_OK;

652

}

653

654

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

655

{

656

if (size > 0) {

657

size--;

658

while (*from && size--)

659

*to++ = *from++;

660

*to = 0;

661

}

662

}

663

664

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

665

{

666

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

667

strcpy(size, to, from);

668

}

669

670

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

671

{

672

char buffer[20];

673

674

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

675

strcat(size, to, buffer);

676

}

677

678

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

679

char *msg;

680

681

result->mr_code = err;

682

*result->mr_stack = 0;

683

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

684

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

685

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

686

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

687

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

688

689

#ifdef XT_WIN

690

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

691

char *ptr;

692

693

ptr = &iMessage[strlen(iMessage)];

694

while (ptr-1 > err_msg) {

695

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

696

break;

697

ptr--;

698

}

699

*ptr = 0;

700

701

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

702

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

703

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

704

return MS_ERR_ENGINE;

705

}

706

#endif

707

708

msg = strerror(err);

709

if (msg) {

710

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

711

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

712

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

713

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

714

}

715

else {

716

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

717

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

718

}

719

720

return MS_ERR_ENGINE;

721

}

722

723

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

724

{

725

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

726

}

727

728

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

729

{

730

while (*cstr && *w_cstr) {

731

if (*cstr != *w_cstr)

732

return false;

733

cstr++;

734

w_cstr++;

735

}

736

return *cstr || !*w_cstr;

737

}

738

739

void deleteTempFiles()

740

{

741

struct dirent entry;

742

struct dirent *result;

743

DIR *odir;

744

int err;

745

char temp_file[100];

746

747

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

748

return;

749

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

750

while (!err && result) {

751

const char **prefix = temp_prefix;

752

753

while (*prefix) {

754

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

755

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

756

757

/* If the process does not exist: */

758

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

759

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

760

unlink(temp_file);

761

}

762

}

763

prefix++;

764

}

765

766

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

767

}

768

closedir(odir);

769

}

770

};

771

#endif // PBMS_API

772

773

774

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

775

776

777

778

779

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

780

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

781

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

782

* done using the direct blob access API

783

784

785

786

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

787

788

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

789

extern void PBMSDeinitBlobStreamingThread(void *v_bs_thread);

790

791

792

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

793

794

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

795

796

797

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

798

799

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

800

801

802

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

803

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

804

805

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

806

807

808

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

809

810

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

811

812

813

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

814

815

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

816

817

818

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

819

820

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

821

#endif //DRIZZLED

822

823

824

#endif //__PBMS_H__

Older »