// case1
INSERT INTO T(v) VALUES(1), (2), (3), (4), (5)
// case2
for i=1; i<=5; i++ {
    INSERT INTO T(v) VALUES(i);
}

${rocketmq.home}\store\commitlog\${fileName}
fileName[n] = fileName[n-1] + mappedFileSize
为了保证mappedFileSize相同，在每个file tail加padding，默认1GB

QueueOffset针对普通消息，存的是consume log里的offset；如果事务消息，是事务状态表的offset
+---------+-------+-----+---------+------+-------------+----------------+----------------+
| MsgSize | Magic | CRC | QueueID | Flag | QueueOffset | PhysicalOffset | SysFlag(P/C/R) |
+---------+-------+-----+---------+------+-------------+----------------+----------------+
+--------------+------------------+-----------+---------------+----+------+-------+------+
| ProducedTime | ProduderHostPort | StoreTime | StoreHostPort | .. | Body | Topic | Prop |
+--------------+------------------+-----------+---------------+----+------+-------+------+

new DispatchRequest(topic, queueId, 
    result.getWroteOffset(), result.getWroteBytes(),
    tagsCode, msg.getStoreTimestamp(), 
    result.getLogicsOffset(), msg.getKeys(),
    // Transaction
    msg.getSysFlag(),
    msg.getPreparedTransactionOffset());

broker1: queue0, queue2
broker2: queue0, 
then, topicA has 3 queues:
broker1_queue0, broker1_queue2, broker2_queue0
producer.selectOneMessageQueue("topicA", "broker1", "queue0")

${rocketmq.home}/store/consumequeue/${topicName}/${queueId}/${fileName}

+---------------------+-----------------+------------------------+
| CommitLogOffset(8B) | MessageSize(4B) | MessageTagHashcode(8B) |
+---------------------+-----------------+------------------------+

TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic()); // from local cache or name server
MessageQueue mq = topicPublishInfo.selectOneMessageQueue(lastBrokerName);
sendResult = this.sendKernelImpl(msg, mq, communicationMode, sendCallback, timeout);

// 2PC，2 messages
// Phase1
producer group write redolog
producer group send a message(type=TransactionPreparedType) to broker
broker append it to CommitLog and return MessageId
broker will not append it to consume queue
// Phase2
producer group write redolog
producer group send a message(type=TransactionCommitType, msgId=$msgId) to broker
broker find the message with msgId in CommitLog and clone it and append it to CommitLog(type=TransactionCommitType|TransactionRollbackType)
if type == TransactionCommitType {
    broker append commit log offset to consume queue
}

24B, mmap
+-----------------+------+-----------+-----------------------+--------------+
| CommitLogOffset | Size | Timestamp | ProducerGroupHashcode | State(P/C/R) |
+-----------------+------+-----------+-----------------------+--------------+
prepare消息，insert table
commit/rollback消息，update table

_, stat, err := cg.kz.conn.Get(path)
if cg.lastVer == -1 {
    // 第一次commit offset
    cg.lastVer = stat.Version
} else if cg.lastVer != stat.Version {
    // user manually reset the offset checkpoint
    return ErrRestartConsumerGroup
}
// 也可能在Get后，用户恰好操作了“回放”，通过CAS解决这个问题
switch err {
case zk.ErrNoNode:
    return cg.kz.create(path, data, false)
case nil: 
    newStat, err := cg.kz.conn.Set(path, data, stat.Version)
    if err != nil {
        cg.lastVer = newStat.Version
    }
    return err
default:
    return err
}

After a crash, we find the following log:
0   BEGIN CHECKPOINT
5   END CHECKPOINT (EMPTY XACT TABLE AND DPT)
10  T1: UPDATE P1 (OLD: YYY NEW: ZZZ)
15  T1: UPDATE P2 (OLD: WWW NEW: XXX)
20  T2: UPDATE P3 (OLD: UUU NEW: VVV)
25  T1: COMMIT
30  T2: UPDATE P1 (OLD: ZZZ NEW: TTT)
Analysis phase:
Scan forward through the log starting at LSN 0.
LSN 5: Initialize XACT table and DPT to empty.
LSN 10: Add (T1, LSN 10) to XACT table. Add (P1, LSN 10) to DPT.
LSN 15: Set LastLSN=15 for T1 in XACT table. Add (P2, LSN 15) to DPT.
LSN 20: Add (T2, LSN 20) to XACT table. Add (P3, LSN 20) to DPT.
LSN 25: Change T1 status to "Commit" in XACT table
LSN 30: Set LastLSN=30 for T2 in XACT table.
Redo phase:
Scan forward through the log starting at LSN 10.
LSN 10: Read page P1, check PageLSN stored in the page. If PageLSN<10, redo LSN 10 (set value to ZZZ) and set the page's PageLSN=10.
LSN 15: Read page P2, check PageLSN stored in the page. If PageLSN<15, redo LSN 15 (set value to XXX) and set the page's PageLSN=15.
LSN 20: Read page P3, check PageLSN stored in the page. If PageLSN<20, redo LSN 20 (set value to VVV) and set the page's PageLSN=20.
LSN 30: Read page P1 if it has been flushed, check PageLSN stored in the page. It will be 10. Redo LSN 30 (set value to TTT) and set the page's PageLSN=30.
Undo phase:
T2 must be undone. Put LSN 30 in ToUndo.
Write Abort record to log for T2
LSN 30: Undo LSN 30 - write a CLR for P1 with "set P1=ZZZ" and undonextLSN=20. Write ZZZ into P1. Put LSN 20 in ToUndo.
LSN 20: Undo LSN 20 - write a CLR for P3 with "set P3=UUU" and undonextLSN=NULL. Write UUU into P3.

// sender
利用OS的API获取screenshot，并以一定的FPS来进行发送
// 优化：把屏幕分成chunk，在把timer之间有变化的chunk生成frame
发送时，frame被编码成H.264或VP8
通过HTTPS发送
// receiver
对接收到的frame解码并显示

<body>
  <p><input type="button" id="share" value="Screen share" /></p>
  <p><video id="video" autoplay /></p>
</body>
<script>
navigator.getUserMedia = navigator.webkitGetUserMedia || navigator.getUserMedia;
$('#share').click(function() {
    navigator.getUserMedia({
         audio: false ,
         video: {
            mandatory: {
                chromeMediaSource: 'screen' ,
                maxWidth: 1280 ,
                maxHeight: 720
            } ,
            optional: [ ]
         }
    }, function(stream) {
       // we've got media stream
       // so the received stream can be transmitted via WebRTC the same way as web camera and easily played in <video> component on the other side
       document.getElementById('video').src = window.URL.createObjectURL(stream);
    } , function() {
       alert('Error. Try in latest Chrome with Screen sharing enabled in about:flags.');
    })
})
</script>

client.send(messageType, incremental, x, y, width, height) => server
// incremental>0，表示该区域内容变化了才发给client；没有变化，就不用发
server.reply(messageType, rectangleN, [{x, y, with, height, color}, ...]) => client

+-------------------+-------------------------+------------+
| index block(16KB) | bloom-filter block(4KB) | data block |
+-------------------+-------------------------+------------+

locate key in memtable
if found then return
locate key in immutable memtable
if found then return
for level:=0; level<=6; level++ {
    // 对于level0，需要遍历所有SSTable，因为keys overlap
    // 但leveldb为了解决这个问题，除了Bloomfilter，也限制了L0的文件数量，一旦超过8，就compact(L0->L1)
    // 对于其他level，由于已经sorted，可以直接定位SSTable
    //
    // 最坏情况下，Get(key)需要读8个L0以及L1-L6，共14个文件
    locate key in SSTable in $level 
    if found then return
}