mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2024-01-19 02:48:24 +00:00
log messages
This commit is contained in:
parent
8c73410a51
commit
f9b6178b8f
|
@ -133,6 +133,7 @@ func ReadEachLogEntry(r io.Reader, sizeBuf []byte, ns int64, eachLogEntryFn func
|
||||||
return fmt.Errorf("size %d bytes, expected 4 bytes", n)
|
return fmt.Errorf("size %d bytes, expected 4 bytes", n)
|
||||||
}
|
}
|
||||||
size := util.BytesToUint32(sizeBuf)
|
size := util.BytesToUint32(sizeBuf)
|
||||||
|
// println("entry size", size)
|
||||||
entryData := make([]byte, size)
|
entryData := make([]byte, size)
|
||||||
n, err = r.Read(entryData)
|
n, err = r.Read(entryData)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|
|
@ -22,7 +22,7 @@ func (broker *MessageBroker) Publish(stream messaging_pb.SeaweedMessaging_Publis
|
||||||
|
|
||||||
// TODO look it up
|
// TODO look it up
|
||||||
topicConfig := &messaging_pb.TopicConfiguration{
|
topicConfig := &messaging_pb.TopicConfiguration{
|
||||||
IsTransient: true,
|
// IsTransient: true,
|
||||||
}
|
}
|
||||||
|
|
||||||
// send init response
|
// send init response
|
||||||
|
|
|
@ -34,7 +34,7 @@ func (broker *MessageBroker) Subscribe(stream messaging_pb.SeaweedMessaging_Subs
|
||||||
|
|
||||||
// TODO look it up
|
// TODO look it up
|
||||||
topicConfig := &messaging_pb.TopicConfiguration{
|
topicConfig := &messaging_pb.TopicConfiguration{
|
||||||
IsTransient: true,
|
// IsTransient: true,
|
||||||
}
|
}
|
||||||
|
|
||||||
if err = stream.Send(&messaging_pb.BrokerMessage{
|
if err = stream.Send(&messaging_pb.BrokerMessage{
|
||||||
|
@ -79,7 +79,7 @@ func (broker *MessageBroker) Subscribe(stream messaging_pb.SeaweedMessaging_Subs
|
||||||
glog.Errorf("unexpected unmarshal messaging_pb.Message: %v", err)
|
glog.Errorf("unexpected unmarshal messaging_pb.Message: %v", err)
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
// fmt.Printf("sending : %d bytes\n", len(m.Value))
|
// fmt.Printf("sending : %d bytes ts %d\n", len(m.Value), logEntry.TsNs)
|
||||||
if err = eachMessageFn(m); err != nil {
|
if err = eachMessageFn(m); err != nil {
|
||||||
glog.Errorf("sending %d bytes to %s: %s", len(m.Value), subscriberId, err)
|
glog.Errorf("sending %d bytes to %s: %s", len(m.Value), subscriberId, err)
|
||||||
return err
|
return err
|
||||||
|
@ -115,6 +115,7 @@ func (broker *MessageBroker) readPersistedLogBuffer(tp *TopicPartition, startTim
|
||||||
startTsNs := startTime.UnixNano()
|
startTsNs := startTime.UnixNano()
|
||||||
|
|
||||||
topicDir := fmt.Sprintf("/topics/%s/%s", tp.Namespace, tp.Topic)
|
topicDir := fmt.Sprintf("/topics/%s/%s", tp.Namespace, tp.Topic)
|
||||||
|
partitionSuffix := fmt.Sprintf(".part%02d", tp.Partition)
|
||||||
|
|
||||||
return filer_pb.List(broker, topicDir, "", func(dayEntry *filer_pb.Entry, isLast bool) error {
|
return filer_pb.List(broker, topicDir, "", func(dayEntry *filer_pb.Entry, isLast bool) error {
|
||||||
dayDir := fmt.Sprintf("%s/%s", topicDir, dayEntry.Name)
|
dayDir := fmt.Sprintf("%s/%s", topicDir, dayEntry.Name)
|
||||||
|
@ -124,7 +125,10 @@ func (broker *MessageBroker) readPersistedLogBuffer(tp *TopicPartition, startTim
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
// println("processing", hourMinuteEntry.FullPath)
|
if !strings.HasSuffix(hourMinuteEntry.Name, partitionSuffix){
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
// println("partition", tp.Partition, "processing", dayDir, "/", hourMinuteEntry.Name)
|
||||||
chunkedFileReader := filer2.NewChunkStreamReader(broker, hourMinuteEntry.Chunks)
|
chunkedFileReader := filer2.NewChunkStreamReader(broker, hourMinuteEntry.Chunks)
|
||||||
defer chunkedFileReader.Close()
|
defer chunkedFileReader.Close()
|
||||||
if err := filer2.ReadEachLogEntry(chunkedFileReader, sizeBuf, startTsNs, eachLogEntryFn); err != nil {
|
if err := filer2.ReadEachLogEntry(chunkedFileReader, sizeBuf, startTsNs, eachLogEntryFn); err != nil {
|
||||||
|
|
|
@ -42,7 +42,7 @@ func (locks *TopicLocks) buildLogBuffer(tl *TopicLock, tp TopicPartition, topicC
|
||||||
flushFn := func(startTime, stopTime time.Time, buf []byte) {
|
flushFn := func(startTime, stopTime time.Time, buf []byte) {
|
||||||
|
|
||||||
if topicConfig.IsTransient {
|
if topicConfig.IsTransient {
|
||||||
return
|
// return
|
||||||
}
|
}
|
||||||
|
|
||||||
// fmt.Printf("flushing with topic config %+v\n", topicConfig)
|
// fmt.Printf("flushing with topic config %+v\n", topicConfig)
|
||||||
|
|
|
@ -93,7 +93,7 @@ func (m *LogBuffer) AddToBuffer(partitionKey, data []byte) {
|
||||||
copy(m.buf[m.pos+4:m.pos+4+size], logEntryData)
|
copy(m.buf[m.pos+4:m.pos+4+size], logEntryData)
|
||||||
m.pos += size + 4
|
m.pos += size + 4
|
||||||
|
|
||||||
// fmt.Printf("entry size %d total %d count %d\n", size, m.pos, len(m.idx))
|
// fmt.Printf("entry size %d total %d count %d, buffer:%p\n", size, m.pos, len(m.idx), m)
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -112,6 +112,7 @@ func (m *LogBuffer) Shutdown() {
|
||||||
func (m *LogBuffer) loopFlush() {
|
func (m *LogBuffer) loopFlush() {
|
||||||
for d := range m.flushChan {
|
for d := range m.flushChan {
|
||||||
if d != nil {
|
if d != nil {
|
||||||
|
// fmt.Printf("flush [%v, %v] size %d\n", d.startTime, d.stopTime, len(d.data.Bytes()))
|
||||||
m.flushFn(d.startTime, d.stopTime, d.data.Bytes())
|
m.flushFn(d.startTime, d.stopTime, d.data.Bytes())
|
||||||
d.releaseMemory()
|
d.releaseMemory()
|
||||||
}
|
}
|
||||||
|
@ -156,7 +157,12 @@ func (m *LogBuffer) ReadFromBuffer(lastReadTime time.Time) (bufferCopy *bytes.Bu
|
||||||
m.RLock()
|
m.RLock()
|
||||||
defer m.RUnlock()
|
defer m.RUnlock()
|
||||||
|
|
||||||
// fmt.Printf("read from buffer: %v last stop time: %v\n", lastReadTime.UnixNano(), m.stopTime.UnixNano())
|
/*
|
||||||
|
fmt.Printf("read buffer %p: %v last stop time: [%v,%v], pos %d, entries:%d, prevBufs:%d\n", m, lastReadTime, m.startTime, m.stopTime, m.pos, len(m.idx), len(m.prevBuffers.buffers))
|
||||||
|
for i, prevBuf := range m.prevBuffers.buffers {
|
||||||
|
fmt.Printf(" prev %d : %s\n", i, prevBuf.String())
|
||||||
|
}
|
||||||
|
*/
|
||||||
|
|
||||||
if lastReadTime.Equal(m.stopTime) {
|
if lastReadTime.Equal(m.stopTime) {
|
||||||
return nil
|
return nil
|
||||||
|
@ -173,6 +179,7 @@ func (m *LogBuffer) ReadFromBuffer(lastReadTime time.Time) (bufferCopy *bytes.Bu
|
||||||
// println("return the earliest in memory", buf.startTime.UnixNano())
|
// println("return the earliest in memory", buf.startTime.UnixNano())
|
||||||
return copiedBytes(buf.buf[:buf.size])
|
return copiedBytes(buf.buf[:buf.size])
|
||||||
}
|
}
|
||||||
|
// println("return the", i, "th in memory", buf.startTime.UnixNano())
|
||||||
return copiedBytes(buf.buf[:buf.size])
|
return copiedBytes(buf.buf[:buf.size])
|
||||||
}
|
}
|
||||||
if !buf.startTime.After(lastReadTime) && buf.stopTime.After(lastReadTime) {
|
if !buf.startTime.After(lastReadTime) && buf.stopTime.After(lastReadTime) {
|
||||||
|
@ -181,6 +188,7 @@ func (m *LogBuffer) ReadFromBuffer(lastReadTime time.Time) (bufferCopy *bytes.Bu
|
||||||
return copiedBytes(buf.buf[pos:buf.size])
|
return copiedBytes(buf.buf[pos:buf.size])
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
// println("return the current buf", lastReadTime.UnixNano())
|
||||||
return copiedBytes(m.buf[:m.pos])
|
return copiedBytes(m.buf[:m.pos])
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
Loading…
Reference in a new issue