erl_memory.c

OTP是开放电信平台的简称

/* ``The contents of this file are subject to the Erlang Public License,
 * Version 1.1, (the "License"); you may not use this file except in
 * compliance with the License. You should have received a copy of the
 * Erlang Public License along with this software. If not, it can be
 * retrieved via the world wide web at http://www.erlang.org/.
 * 
 * Software distributed under the License is distributed on an "AS IS"
 * basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
 * the License for the specific language governing rights and limitations
 * under the License.
 * 
 * The Initial Developer of the Original Code is Ericsson Utvecklings AB.
 * Portions created by Ericsson are Copyright 1999, Ericsson Utvecklings
 * AB. All Rights Reserved.''
 * 
 *     $Id$
 */


/*
 * Description:	
 *
 * Author: 	Rickard Green
 */

/* Headers to include ... */

#ifdef __WIN32__
#	include <winsock2.h>
#	undef WIN32_LEAN_AND_MEAN
#	define WIN32_LEAN_AND_MEAN
#	include <windows.h>
#else
#	if defined(__linux__) && defined(__GNUC__)
#   		define _GNU_SOURCE 1
#	endif
#	include <unistd.h>
#	include <sys/types.h>
#	include <sys/socket.h>
#	include <netinet/in.h>
#	include <fcntl.h>
#	include <netdb.h>
#	include <arpa/inet.h>
#endif

#include <stdio.h>
#include <stdlib.h>
#include <limits.h>
#include <string.h>

#include "erl_fixed_size_int_types.h"
#include "erl_memory_trace_parser.h"
#include "erl_memory_trace_block_table.h"
#include "ethread.h"

/* Increment when changes are made */
#define EMEM_VSN_STR "0.9"

/* Features not fully implemented yet */
#define EMEM_A_SWITCH 0
#define EMEM_C_SWITCH 0
#define EMEM_c_SWITCH 0
#define EMEM_d_SWITCH 0

/* Some system specific defines ... */
#ifdef __WIN32__
#	define ssize_t int
#	define GET_SOCK_ERRNO() (WSAGetLastError() - WSABASEERR)
#	define IS_INVALID_SOCKET(X) ((X) == INVALID_SOCKET)
#	ifdef __GNUC__
#		define INLINE __inline__
#	else
#	define INLINE __forceinline
#       endif
#	define DIR_SEP_CHAR '\\'
#else
#	define SOCKET int
#	define closesocket close
#	define GET_SOCK_ERRNO() (errno ? errno : INT_MAX)
#	define INVALID_SOCKET (-1)
#	define IS_INVALID_SOCKET(X) ((X) < 0)
#	ifdef __GNUC__
#		define INLINE __inline__
#	else
#		define INLINE
#	endif
#	define DIR_SEP_CHAR '/'
#endif

#define EM_ERL_CMD_FILE_NAME "erl_cmd.txt"
#define EM_OUTPUT_FILE_SUFFIX ".emem"

#define PRINT_OPERATIONS 0

/* Our own assert() ... */
#ifdef DEBUG
#define ASSERT(A) ((void) ((A) ? 1 : assert_failed(__FILE__, __LINE__, #A)))
#include <stdio.h>
static int assert_failed(char *f, int l, char *a)
{
    fprintf(stderr, "%s:%d: Assertion failed: %s\n", f, l, a);
    abort();
    return 0;
}

#else
#define ASSERT(A) ((void) 1)
#endif

#define ERR_RET(X) return (X)
#if 1
#  undef  ERR_RET
#  define ERR_RET(X) abort()
#endif

/* #define HARD_DEBUG */


#define EM_EXIT_RESULT			(EMTBT_MIN_ERROR - 1)
#define EM_TRUNCATED_TRACE_ERROR	(EMTBT_MIN_ERROR - 2)
#define EM_INTERNAL_ERROR		(EMTBT_MIN_ERROR - 3)

#define EM_DEFAULT_BUF_SZ 8192

#define EM_LINES_UNTIL_HEADER 20
#define EM_NO_OF_OPS 400
#define EM_MAX_CONSECUTIVE_TRACE_READS 10
#define EM_MAX_NO_OF_TRACE_BUFS 1280
#define EM_MIN_TRACE_READ_SIZE (EM_DEFAULT_BUF_SZ/20)
#define EM_TIME_FIELD_WIDTH 11

static void error(int res);
static void error_msg(int res, char *msg);

typedef struct {
    usgnd_int_max size;
    usgnd_int_max min_size;
    usgnd_int_max max_size;
    usgnd_int_max max_ever_size;
    usgnd_int_max no;
    usgnd_int_max min_no;
    usgnd_int_max max_no;
    usgnd_int_max max_ever_no;
    usgnd_int_max allocs;
    usgnd_int_max reallocs;
    usgnd_int_max frees;
} em_mem_info;

typedef struct em_buffer_ {
    struct em_buffer_ *next;
    int write;
    usgnd_int_8 *data;
    usgnd_int_8 *data_end;
    usgnd_int_8 *end;
    size_t size;
    usgnd_int_8 start[EM_DEFAULT_BUF_SZ];
} em_buffer;

typedef struct {
    int no_writer;
    int no_reader;
    size_t tot_buf_size;
    size_t max_buf_size;
    char *name;
    em_buffer *first;
    em_buffer *last;
    ethr_mutex mutex;
    ethr_cond cond;
    int used_def_buf_a;
    em_buffer def_buf_a;
    int used_def_buf_b;
    em_buffer def_buf_b;
} em_buf_queue;

typedef struct {
    usgnd_int_8 *ptr;
    size_t size;
} em_area;

typedef struct {
    char *name;
    int ix;
} em_output_types;

typedef struct {

    /* Memory allocation functions */
    void * (*alloc)(size_t);
    void * (*realloc)(void *, size_t);
    void   (*free)(void *);

    emtbt_table *block_table;
    emtbt_table **carrier_table;

    struct {
	em_mem_info total;
	em_mem_info *btype;
	em_mem_info *allctr;
	em_mem_info **allctr_prv_crr;
	em_mem_info **allctr_usd_crr;

	struct {
	    usgnd_int_32 secs;
	    usgnd_int_32 usecs;
	} stop_time;
	emtp_op_type stop_reason;
	usgnd_int_32 exit_status;
    } info;

    /* Input ... */
    struct {
	usgnd_int_16 listen_port;
	SOCKET socket;
	usgnd_int_max total_trace_size;
	int error;
	char *error_descr;
	em_buf_queue queue;
    } input;

    /* Output ... */
    struct {
	usgnd_int_32 next_print;
	usgnd_int_32 next_print_inc;
	char *header;
	size_t header_size;
	size_t values_per_object;
	size_t values_per_line;
	size_t field_width;
	int verbose;
	int total;
	int all_allctrs;
	int no_allctrs;
	em_output_types *allctrs;
	int all_btypes;
	int no_btypes;
	em_output_types *btypes;
	int max_min_values;
	int block_counts;
	int op_counts;
	int lines_until_header;
	FILE *stream;
	char *file_name;
#if EMEM_d_SWITCH
	char *dir_name;
	FILE *erl_cmd_file;
	struct {
	    ethr_mutex *mutex;
	    ethr_cond *cond;
	} go;
#endif
	em_buf_queue queue;
    } output;

    /* Trace info */
    emtp_state *trace_state;
    emtp_info  trace_info;

} em_state;

/* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * *\
 * Threads...                                                              *
 *                                                                         *
\*                                                                         */

static INLINE void
mutex_init(ethr_mutex *mtx)
{
    int res = ethr_mutex_init(mtx);
    if (res)
	error_msg(res, "Mutex init");
}

static INLINE void
mutex_destroy(ethr_mutex *mtx)
{
    int res = ethr_mutex_destroy(mtx);
    if (res)
	error_msg(res, "Mutex destroy");
}

static INLINE void
mutex_lock(ethr_mutex *mtx)
{
    int res = ethr_mutex_lock(mtx);
    if (res)
	error_msg(res, "Mutex lock");
}

static INLINE void
mutex_unlock(ethr_mutex *mtx)
{
    int res = ethr_mutex_unlock(mtx);
    if (res)
	error_msg(res, "Mutex unlock");
}

static INLINE void
cond_init(ethr_cond *cnd)
{
    int res = ethr_cond_init(cnd);
    if (res)
	error_msg(res, "Cond init");
}

static INLINE void
cond_destroy(ethr_cond *cnd)
{
    int res = ethr_cond_destroy(cnd);
    if (res)
	error_msg(res, "Cond destroy");
}

static INLINE void
cond_wait(ethr_cond *cnd, ethr_mutex *mtx)
{
    int res = ethr_cond_wait(cnd, mtx);
    if (res)
	error_msg(res, "Cond wait");
}

static INLINE void
cond_signal(ethr_cond *cnd)
{
    int res = ethr_cond_signal(cnd);
    if (res)
	error_msg(res, "Cond signal");
}


/* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * *\
 * Buffer queues                                                           *
 *                                                                         *
\*                                                                         */

static INLINE void
reset_buffer(em_buffer *b, size_t size)
{
    b->write = 1;
    b->next = NULL;
    if (size) {
	b->size = size;
	b->end = b->start + size;
    }
    b->data_end = b->data = b->start;
}

static void
init_queue(em_state *state, em_buf_queue *queue)
{
    reset_buffer(&queue->def_buf_a, EM_DEFAULT_BUF_SZ);
    reset_buffer(&queue->def_buf_b, EM_DEFAULT_BUF_SZ);
    queue->first = NULL;
    queue->last = NULL;
    queue->no_writer = 0;
    queue->no_reader = 0;
    queue->tot_buf_size = 0;
    queue->max_buf_size = ~0;
    queue->name = "";
    queue->used_def_buf_a = 0;
    queue->used_def_buf_b = 0;
    mutex_init(&queue->mutex);
    cond_init(&queue->cond);
}

static void
destroy_queue(em_state *state, em_buf_queue *queue)
{
    while (queue->first) {
	em_buffer *buf = queue->first;
	queue->first = queue->first->next;
	if (buf != &queue->def_buf_a && buf != &queue->def_buf_b)
	    (*state->free)((void *) buf);
    }
    mutex_destroy(&queue->mutex);
    cond_destroy(&queue->cond);
}

static void
disconnect_queue_writer(em_buf_queue *queue)
{
    mutex_lock(&queue->mutex);
    queue->no_writer = 1;
    cond_signal(&queue->cond);
    mutex_unlock(&queue->mutex);
}

static void
disconnect_queue_reader(em_buf_queue *queue)
{
    mutex_lock(&queue->mutex);
    queue->no_reader = 1;
    cond_signal(&queue->cond);
    mutex_unlock(&queue->mutex);
}

static int
is_queue_writer_disconnected(em_buf_queue *queue)
{
    int res;
    mutex_lock(&queue->mutex);
    res = queue->no_writer;
    mutex_unlock(&queue->mutex);
    return res;
}

static int
is_queue_reader_disconnected(em_buf_queue *queue)
{
    int res;
    mutex_lock(&queue->mutex);
    res = queue->no_reader;
    mutex_unlock(&queue->mutex);
    return res;
}

static INLINE void
dequeue(em_state *state, em_buf_queue *queue)
{
    em_buffer *buf;

    ASSERT(queue->first);
    ASSERT(queue->tot_buf_size > 0);

    buf = queue->first;
    queue->first = buf->next;
    if (!queue->first)
	queue->last = NULL;

    ASSERT(queue->tot_buf_size >= buf->size);
    queue->tot_buf_size -= buf->size;

    if (buf == &queue->def_buf_a)
	queue->used_def_buf_a = 0;
    else if (buf == &queue->def_buf_b)
	queue->used_def_buf_b = 0;
    else
	(*state->free)((void *) buf);

}


static INLINE em_buffer *
enqueue(em_state *state, em_buf_queue *queue, size_t min_size)
{
    em_buffer *buf;

    if (min_size > EM_DEFAULT_BUF_SZ)
	goto alloc_buf;

    if (!queue->used_def_buf_a) {
	buf = &queue->def_buf_a;
	queue->used_def_buf_a = 1;
	reset_buffer(buf, 0);
    }
    else if (!queue->used_def_buf_b) {
	buf = &queue->def_buf_b;
	queue->used_def_buf_b = 1;
	reset_buffer(buf, 0);
    }
    else {
	size_t bsize;
    alloc_buf:

	bsize = EM_DEFAULT_BUF_SZ;
	if (bsize < min_size)
	    bsize = min_size;
	
	buf = (em_buffer *) (*state->alloc)(sizeof(em_buffer)
					    + (sizeof(usgnd_int_8)
					       * (bsize-EM_DEFAULT_BUF_SZ)));
	if (buf) {
	    buf->size = bsize;
	    reset_buffer(buf, bsize);
	}
    }

    if (queue->last) {
	ASSERT(queue->first);
	queue->last->write = 0;
	queue->last->next = buf;
    }
    else {
	ASSERT(!queue->first);
	queue->first = buf;
    }

    queue->tot_buf_size += buf->size;
    queue->last = buf;

    return buf;
}

static void
get_next_read_area(em_area *area, em_state *state, em_buf_queue *queue)
{
    mutex_lock(&queue->mutex);

    while (!queue->first || queue->first->data == queue->first->data_end) {
	if (queue->first && (!queue->first->write
			     || queue->first->data == queue->first->end)) {
	    dequeue(state, queue);
	    continue;
	}

	if (queue->no_writer) {
	    area->ptr = NULL;
	    area->size = 0;
	    mutex_unlock(&queue->mutex);
	    return;
	}
	cond_wait(&queue->cond, &queue->mutex);
    }

    ASSERT(queue->first->data < queue->first->data_end);

    area->ptr = queue->first->data;
    area->size =  queue->first->data_end - queue->first->data;

    queue->first->data = queue->first->data_end;

    mutex_unlock(&queue->mutex);