qemu-devel
[Top][All Lists]
Advanced

[Date Prev][Date Next][Thread Prev][Thread Next][Date Index][Thread Index]

Re: [Qemu-devel] [PATCH v4 2/3] Simpletrace v2: Support multiple argumen


From: Stefan Hajnoczi
Subject: Re: [Qemu-devel] [PATCH v4 2/3] Simpletrace v2: Support multiple arguments, strings.
Date: Wed, 13 Jun 2012 13:25:46 +0100

On Tue, Jun 12, 2012 at 4:16 PM, Harsh Prateek Bora
<address@hidden> wrote:
> +int trace_alloc_record(TraceBufferRecord *rec, TraceEventID event, uint32_t 
> datasize);

This should be static.

> +static void read_from_buffer(unsigned int idx, void *dataptr, size_t size);
> +static unsigned int write_to_buffer(unsigned int idx, void *dataptr, size_t 
> size);
> +void trace_mark_record_complete(TraceBufferRecord *rec);

This should be static.  Can we get rid of these internal names
completely?  trace_record_finish() could hold the code instead of
calling into an internal function.  Same for trace_record_start()
above.

> +
>  /**
>  * Read a trace record from the trace buffer
>  *
> @@ -75,16 +86,27 @@ static char *trace_file_name = NULL;
>  *
>  * Returns false if the record is not valid.
>  */
> -static bool get_trace_record(unsigned int idx, TraceRecord *record)
> +static bool get_trace_record(unsigned int idx, TraceRecord **recordptr)
>  {
> -    if (!(trace_buf[idx].event & TRACE_RECORD_VALID)) {
> +    uint8_t rec_hdr[sizeof(TraceRecord)];
> +    uint64_t event_flag = 0;
> +    TraceRecord *record = (TraceRecord *) rec_hdr;
> +    /* read the event flag to see if its a valid record */
> +    read_from_buffer(idx, rec_hdr, sizeof(event_flag));
> +
> +    if (!(record->event & TRACE_RECORD_VALID)) {
>         return false;
>     }
>
>     __sync_synchronize(); /* read memory barrier before accessing record */
> -
> -    *record = trace_buf[idx];
> -    record->event &= ~TRACE_RECORD_VALID;
> +    /* read the record header to know record length */
> +    read_from_buffer(idx, rec_hdr, sizeof(TraceRecord));
> +    *recordptr = g_malloc(record->length);
> +    /* make a copy of record to avoid being overwritten */
> +    read_from_buffer(idx, (uint8_t *)*recordptr, record->length);
> +    (*recordptr)->event &= ~TRACE_RECORD_VALID;

Here I think we need another barrier - we don't want to read data
after clearing the valid flag (otherwise we might read data for a new
event, not the event we just read!).  This is being superparanoid but
let's be careful for starters and if someone can explain why it's safe
to eliminate the barrier, then we can remove it.

> +    /* reset record event validity flag on global trace buffer */
> +    write_to_buffer(idx, &event_flag, sizeof(event_flag));
>     return true;
>  }
>
> @@ -120,29 +142,30 @@ static void wait_for_trace_records_available(void)
>
>  static gpointer writeout_thread(gpointer opaque)
>  {
> -    TraceRecord record;
> -    unsigned int writeout_idx = 0;
> -    unsigned int num_available, idx;
> +    TraceRecord *recordptr, *dropped_ptr;
> +    uint8_t dropped_rec[sizeof(TraceRecord) + sizeof(dropped_events)];
> +    dropped_ptr = (TraceRecord *)dropped_rec;

Since you need to resend, please use a union instead of a uint8_t
array and a pointer.  This tells the compiler your intention and
guarantees that the memory is aligned properly on all host platforms.

union {
    TraceRecord rec;
    uint8_t bytes[sizeof(TraceRecord) + sizeof(uint64_t)];
} dropped;

> +    unsigned int idx = 0;
>     size_t unused __attribute__ ((unused));
>
>     for (;;) {
>         wait_for_trace_records_available();
>
> -        num_available = trace_idx - writeout_idx;
> -        if (num_available > TRACE_BUF_LEN) {
> -            record = (TraceRecord){
> -                .event = DROPPED_EVENT_ID,
> -                .x1 = num_available,
> -            };
> -            unused = fwrite(&record, sizeof(record), 1, trace_fp);
> -            writeout_idx += num_available;
> +        if (dropped_events) {
> +            dropped_ptr->event = DROPPED_EVENT_ID,
> +            dropped_ptr->timestamp_ns = get_clock();
> +            dropped_ptr->length = sizeof(TraceRecord) + 
> sizeof(dropped_events),
> +            dropped_ptr->reserved = 0;
> +            memcpy(dropped_ptr->arguments, &dropped_events, 
> sizeof(dropped_events));
> +            dropped_events = 0;

To avoid losing counts we need cmpxchg &dropped_events, 0.  Remember
one or more threads might be writing to dropped_events while this code
executes!  Since this value is a count, not a boolean, let's ensure it
is accurate.

> +            unused = fwrite(dropped_ptr, dropped_ptr->length, 1, trace_fp);
>         }
>
> -        idx = writeout_idx % TRACE_BUF_LEN;
> -        while (get_trace_record(idx, &record)) {
> -            trace_buf[idx].event = 0; /* clear valid bit */
> -            unused = fwrite(&record, sizeof(record), 1, trace_fp);
> -            idx = ++writeout_idx % TRACE_BUF_LEN;
> +        while (get_trace_record(idx, &recordptr)) {
> +            unused = fwrite(recordptr, recordptr->length, 1, trace_fp);
> +            writeout_idx += recordptr->length;
> +            g_free(recordptr);
> +            idx = writeout_idx % TRACE_BUF_LEN;
>         }

This has become quite nice!

> -void trace3(TraceEventID event, uint64_t x1, uint64_t x2, uint64_t x3)
> +int trace_alloc_record(TraceBufferRecord *rec, TraceEventID event, uint32_t 
> datasize)
>  {
> -    trace(event, x1, x2, x3, 0, 0, 0);
> +    unsigned int idx, rec_off, old_idx, new_idx;
> +    uint32_t rec_len = sizeof(TraceRecord) + datasize;
> +    uint64_t timestamp_ns = get_clock();
> +
> +    while (1) {
> +        old_idx = trace_idx;
> +        smp_rmb();
> +        new_idx = old_idx + rec_len;
> +
> +        if (new_idx - writeout_idx > TRACE_BUF_LEN) {
> +            dropped_events++; /* Trace Buffer Full, Event dropped ! */

See comment above about making dropped_events atomic.

> +            return - ENOSPC;

Just a syntax nitpick but it's very unusual to put a space after the
negative, it looks like you were trying to use the binary minus
operator.  The usual way is "-ENOSPC".

> +        }
> +
> +        if g_atomic_int_compare_and_exchange((gint *)&trace_idx, old_idx, 
> new_idx)
> +            break;

Please run scripts/checkpatch.pl.  QEMU always uses {} for bodies of
if/while statements.

> +    }
> +
> +    idx = old_idx % TRACE_BUF_LEN;
> +    /*  To check later if threshold crossed */
> +    rec->next_tbuf_idx = trace_idx % TRACE_BUF_LEN;

Careful, this looks wrong since trace_idx can change while this
function is executing.  Did you mean new_idx?

> -void trace6(TraceEventID event, uint64_t x1, uint64_t x2, uint64_t x3, 
> uint64_t x4, uint64_t x5, uint64_t x6)
> +void trace_mark_record_complete(TraceBufferRecord *rec)
>  {
> -    trace(event, x1, x2, x3, x4, x5, x6);
> +    uint8_t temp_rec[sizeof(TraceRecord)];
> +    TraceRecord *record = (TraceRecord *) temp_rec;
> +    read_from_buffer(rec->tbuf_idx, temp_rec, sizeof(TraceRecord));
> +    __sync_synchronize(); /* write barrier before marking as valid */
> +    record->event |= TRACE_RECORD_VALID;
> +    write_to_buffer(rec->tbuf_idx, temp_rec, sizeof(TraceRecord));
> +
> +    if (rec->next_tbuf_idx > TRACE_BUF_FLUSH_THRESHOLD &&
> +        rec->tbuf_idx <= TRACE_BUF_FLUSH_THRESHOLD) {
> +        flush_trace_file(false);
> +    }

There is an important difference between this new code and the old
code.  Previously we did modulo TRACE_BUF_FLUSH_THRESHOLD, this cause
a flush *every* TRACE_BUF_FLUSH_THRESHOLD entries.  The new code only
flushes once: when we cross the TRACE_BUF_FLUSH_THRESHOLD byte
boundary.

This new flush condition is no better than flushing when we wrap the
buffer.  In other words, the buffer will be full when we flush!  We
need to avoid this so that the chance of events getting dropped is
low.

The idea of checking when we cross the threshold is good, but it needs
to apply *every* TRACE_BUF_FLUSH_THRESHOLD bytes, not just once.

Stefan



reply via email to

[Prev in Thread] Current Thread [Next in Thread]