path: root/qpid/cpp/src/qpid/store/ms-clfs/MessageLog.cpp

/*
 *
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 * 
 *   http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 *
 */

#include <windows.h>
#include <clfsw32.h>
#include <exception>
#include <malloc.h>
#include <memory.h>
#include <qpid/framing/Buffer.h>
#include <qpid/log/Statement.h>
#include <qpid/sys/IntegerTypes.h>
#include <qpid/sys/windows/check.h>

#include "MessageLog.h"
#include "Lsn.h"

namespace qpid {
namespace store {
namespace ms_clfs {

namespace {

// Structures that hold log records. Each has a type field at the start.
enum MessageEntryType {
    MessageStartEntry           = 1,
    MessageChunkEntry           = 2,
    MessageDeleteEntry          = 3,
    MessageEnqueueEntry         = 4,
    MessageDequeueEntry         = 5
};
static const uint32_t MaxMessageContentLength = 64 * 1024;

// Message-Start
struct MessageStart {
    MessageEntryType type;
    // If the complete message encoding doesn't fit, remainder is in
    // MessageChunk records to follow.
    // headerLength is the size of the message's header in content. It is
    // part of the totalLength and the segmentLength.
    uint32_t headerLength;
    uint32_t totalLength;
    uint32_t segmentLength;
    char content[MaxMessageContentLength];

    MessageStart()
      : type(MessageStartEntry),
        headerLength(0),
        totalLength(0),
        segmentLength(0) {}
};
// Message-Chunk
struct MessageChunk {
    MessageEntryType type;
    uint32_t segmentLength;
    char content[MaxMessageContentLength];

    MessageChunk() : type(MessageChunkEntry), segmentLength(0) {}
};
// Message-Delete
struct MessageDelete {
    MessageEntryType type;

    MessageDelete() : type(MessageDeleteEntry) {}
};
// Message-Enqueue
struct MessageEnqueue {
    MessageEntryType type;
    uint64_t queueId;
    uint64_t transId;

    MessageEnqueue(uint64_t qId = 0, uint64_t tId = 0)
        : type(MessageEnqueueEntry), queueId(qId), transId(tId) {}
};
// Message-Dequeue
struct MessageDequeue {
    MessageEntryType type;
    uint64_t queueId;
    uint64_t transId;

    MessageDequeue(uint64_t qId = 0, uint64_t tId = 0)
        : type(MessageDequeueEntry), queueId(qId), transId(tId) {}
};

}   // namespace

void
MessageLog::initialize()
{
    // Write something to occupy the first record, preventing a real message
    // from being lsn/id 0. Delete of a non-existant id is easily tossed
    // during recovery if no other messages have caused the tail to be moved
    // up past this dummy record by then.
    deleteMessage(0, 0);
}

uint32_t
MessageLog::marshallingBufferSize()
{
    size_t biggestNeed = std::max(sizeof(MessageStart), sizeof(MessageEnqueue));
    uint32_t defSize = static_cast<uint32_t>(biggestNeed);
    uint32_t minSize = Log::marshallingBufferSize();
    if (defSize <= minSize)
        return minSize;
    // Round up to multiple of minSize
    return (defSize + minSize) / minSize * minSize;
}

uint64_t
MessageLog::add(const boost::intrusive_ptr<qpid::broker::PersistableMessage>& msg)
{
    // The message may be too long to fit in one record; if so, write
    // Message-Chunk records to contain the rest. If it does all fit in one
    // record, though, optimize the encoding by going straight to the
    // Message-Start record rather than encoding then copying to the record.
    // In all case
    MessageStart entry;
    uint32_t encodedMessageLength = msg->encodedSize();
    entry.headerLength = msg->encodedHeaderSize();
    entry.totalLength = encodedMessageLength;
    CLFS_LSN location, lastChunkLsn;
    std::auto_ptr<char> encodeStage;
    char *encodeBuff = 0;
    bool oneRecord = encodedMessageLength <= MaxMessageContentLength;
    if (oneRecord) {
        encodeBuff = entry.content;
        entry.segmentLength = encodedMessageLength;
    }
    else {
        encodeStage.reset(new char[encodedMessageLength]);
        encodeBuff = encodeStage.get();
        entry.segmentLength = MaxMessageContentLength;
    }
    qpid::framing::Buffer buff(encodeBuff, encodedMessageLength);
    msg->encode(buff);
    if (!oneRecord)
        memcpy_s(entry.content, sizeof(entry.content),
                 encodeBuff, entry.segmentLength);
    uint32_t entryLength = static_cast<uint32_t>(sizeof(entry));
    entryLength -= (MaxMessageContentLength - entry.segmentLength);
    location = write(&entry, entryLength);
    // Write any Message-Chunk records before setting the message's id.
    uint32_t sent = entry.segmentLength;
    uint32_t remaining = encodedMessageLength - entry.segmentLength;
    while (remaining > 0) {
        MessageChunk chunk;
        chunk.segmentLength = std::max(MaxMessageContentLength, remaining);
        memcpy_s(chunk.content, sizeof(chunk.content),
                 encodeStage.get() + sent, chunk.segmentLength);
        entryLength = static_cast<uint32_t>(sizeof(chunk));
        entryLength -= (MaxMessageContentLength - chunk.segmentLength);
        lastChunkLsn = write(&chunk, entryLength, &location);
        sent += chunk.segmentLength;
        remaining -= chunk.segmentLength;
    }
    return lsnToId(location);
}

void
MessageLog::deleteMessage(uint64_t messageId, uint64_t newFirstId)
{
    MessageDelete deleteEntry;
    CLFS_LSN msgLsn = idToLsn(messageId);
    write(&deleteEntry, sizeof(deleteEntry), &msgLsn);
    if (newFirstId != 0)
        moveTail(idToLsn(newFirstId));
}

// Load part or all of a message's content from previously stored
// log record(s).
void
MessageLog::loadContent(uint64_t messageId,
                        std::string& data,
                        uint64_t offset,
                        uint32_t length)
{
}

void
MessageLog::recordEnqueue (uint64_t messageId,
                           uint64_t queueId,
                           uint64_t transactionId)
{
    MessageEnqueue entry(queueId, transactionId);
    CLFS_LSN msgLsn = idToLsn(messageId);
    write(&entry, sizeof(entry), &msgLsn);
}

void
MessageLog::recordDequeue (uint64_t messageId,
                           uint64_t queueId,
                           uint64_t transactionId)
{
    MessageDequeue entry(queueId, transactionId);
    CLFS_LSN msgLsn = idToLsn(messageId);
    write(&entry, sizeof(entry), &msgLsn);
}

void
MessageLog::recover(qpid::broker::RecoveryManager& recoverer,
                    qpid::store::MessageMap& messageMap,
                    std::map<uint64_t, std::vector<RecoveredMsgOp> >& messageOps)
{
    // If context and content needs to be saved while reassembling messages
    // split across log records, save the info and reassembly buffer.
    struct MessageBlocks {
        uint32_t totalLength;
        uint32_t soFarLength;
        boost::shared_ptr<char> content;

        MessageBlocks() : totalLength(0), soFarLength(0), content((char*)0) {}
    };
    std::map<uint64_t, MessageBlocks> reassemblies;
    std::map<uint64_t, MessageBlocks>::iterator at;

    QPID_LOG(debug, "Recovering message log");

    // Note that there may be message refs in the log which are deleted, so
    // be sure to only add msgs at message-start record, and ignore those
    // that don't have an existing message record.
    // Get the base LSN - that's how to say "start reading at the beginning"
    CLFS_INFORMATION info;
    ULONG infoLength = sizeof (info);
    BOOL ok = ::GetLogFileInformation(handle, &info, &infoLength);
    QPID_WINDOWS_CHECK_NOT(ok, 0);

    // Pointers for the various record types that can be assigned in the
    // reading loop below.
    MessageStart *start;
    MessageChunk *chunk;
    MessageEnqueue *enqueue;
    MessageDequeue *dequeue;

    qpid::store::MessageMap::iterator messageMapSpot;
    qpid::store::MessageQueueMap::iterator queueMapSpot;
    PVOID recordPointer;
    ULONG recordLength;
    CLFS_RECORD_TYPE recordType = ClfsDataRecord;
    CLFS_LSN messageLsn, current, undoNext;
    PVOID readContext;
    uint64_t msgId;
    // Note 'current' in case it's needed below; ReadNextLogRecord returns it
    // via a parameter.
    current = info.BaseLsn;
    ok = ::ReadLogRecord(marshal,
                         &info.BaseLsn,
                         ClfsContextForward,
                         &recordPointer,
                         &recordLength,
                         &recordType,
                         &undoNext,
                         &messageLsn,
                         &readContext,
                         0);
    while (ok) {
        // All the record types this class writes have a MessageEntryType in the
        // beginning. Based on that, do what's needed.
        MessageEntryType *t =
            reinterpret_cast<MessageEntryType *>(recordPointer);
        switch(*t) {
        case MessageStartEntry:
            start = reinterpret_cast<MessageStart *>(recordPointer);
            msgId = lsnToId(current);
            QPID_LOG(debug, "Message Start, id " << msgId);
            // If the message content is split across multiple log records, save
            // this content off to the side until the remaining record(s) are
            // located.
            if (start->totalLength == start->segmentLength) {  // Whole thing
                // Start by recovering the header then see if the rest of
                // the content is desired.
                qpid::framing::Buffer buff(start->content, start->headerLength);
                qpid::broker::RecoverableMessage::shared_ptr m =
                    recoverer.recoverMessage(buff);
                m->setPersistenceId(msgId);
                messageMap[msgId] = m;
                uint32_t contentLength =
                    start->totalLength - start->headerLength;
                if (m->loadContent(contentLength)) {
                    qpid::framing::Buffer content(&(start->content[start->headerLength]),
                                                  contentLength);
                    m->decodeContent(content);
                }
            }
            else {
                // Save it in a block big enough.
                MessageBlocks b;
                b.totalLength = start->totalLength;
                b.soFarLength = start->segmentLength;
                b.content.reset(new char[b.totalLength]);
                memcpy_s(b.content.get(), b.totalLength,
                         start->content, start->segmentLength);
                reassemblies[msgId] = b;
            }
            break;
        case MessageChunkEntry:
            chunk = reinterpret_cast<MessageChunk *>(recordPointer);
            // Remember, all entries chained to MessageStart via previous.
            msgId = lsnToId(messageLsn);
            QPID_LOG(debug, "Message Chunk for id " << msgId);
            at = reassemblies.find(msgId);
            if (at == reassemblies.end()) {
                QPID_LOG(debug, "Message frag for " << msgId <<
                                " but no start; discarded");
            }
            else {
                MessageBlocks *b = &(at->second);
                if (b->soFarLength + chunk->segmentLength > b->totalLength)
                    throw std::runtime_error("Invalid message chunk length");
                memcpy_s(b->content.get() + b->soFarLength,
                         b->totalLength - b->soFarLength,
                         chunk->content,
                         chunk->segmentLength);
                b->soFarLength += chunk->segmentLength;
                if (b->totalLength == b->soFarLength) {
                    qpid::framing::Buffer buff(b->content.get(),
                                               b->totalLength);
                    qpid::broker::RecoverableMessage::shared_ptr m =
                        recoverer.recoverMessage(buff);
                    m->setPersistenceId(msgId);
                    messageMap[msgId] = m;
                    reassemblies.erase(at);
                }
            }
            break;
        case MessageDeleteEntry:
            msgId = lsnToId(messageLsn);
            QPID_LOG(debug, "Message Delete, id " << msgId);
            messageMap.erase(msgId);
            messageOps.erase(msgId);
            break;
        case MessageEnqueueEntry:
            enqueue = reinterpret_cast<MessageEnqueue *>(recordPointer);
            msgId = lsnToId(messageLsn);
            QPID_LOG(debug, "Message " << msgId << " Enqueue on queue " <<
                            enqueue->queueId << ", txn " << enqueue->transId);
            if (messageMap.find(msgId) == messageMap.end()) {
                QPID_LOG(debug,
                         "Message " << msgId << " doesn't exist; discarded");
            }
            else {
                std::vector<RecoveredMsgOp>& ops = messageOps[msgId];
                RecoveredMsgOp op(RECOVERED_ENQUEUE,
                                  enqueue->queueId,
                                  enqueue->transId);
                ops.push_back(op);
            }
            break;
        case MessageDequeueEntry:
            dequeue = reinterpret_cast<MessageDequeue *>(recordPointer);
            msgId = lsnToId(messageLsn);
            QPID_LOG(debug, "Message " << msgId << " Dequeue from queue " <<
                            dequeue->queueId);
            if (messageMap.find(msgId) == messageMap.end()) {
                QPID_LOG(debug,
                         "Message " << msgId << " doesn't exist; discarded");
            }
            else {
                std::vector<RecoveredMsgOp>& ops = messageOps[msgId];
                RecoveredMsgOp op(RECOVERED_DEQUEUE,
                                  dequeue->queueId,
                                  dequeue->transId);
                ops.push_back(op);
            }
            break;
        default:
            throw std::runtime_error("Bad message log entry type");
        }

        recordType = ClfsDataRecord;
        ok = ::ReadNextLogRecord(readContext,
                                 &recordPointer,
                                 &recordLength,
                                 &recordType,
                                 0,             // No userLsn
                                 &undoNext,
                                 &messageLsn,
                                 &current,
                                 0);
    }
    DWORD status = ::GetLastError();
    ::TerminateReadLog(readContext);
    if (status == ERROR_HANDLE_EOF) { // No more records
        QPID_LOG(debug, "Message log recovered");
        return;
    }
    throw QPID_WINDOWS_ERROR(status);
}

}}}  // namespace qpid::store::ms_clfs