char* operation practice

setup.py

# -*- coding: utf-8 -*-

try:
  from setuptools.setuptools import setup, Extension
except:
  from distutils.core import setup, Extension
  
import sys
import os
from subprocess import Popen, PIPE

name = "test"
version = "1.0"
short_description = "test module"
long_description = "this is test module"

define_macros = []
libraries = []
library_dirs = []
include_dirs = []

test_module = Extension('test',
                        sources = ["testmodule.c"],
                        libraries = libraries,
                        library_dirs = library_dirs,
                        include_dirs = include_dirs,
                        define_macros = define_macros)

# setup
setup(name = name,
      version = version,
      description = short_description,
      long_description = long_description,
      author = "Yoshifumi YAMAGUCHI (ymotongpoo)",
      author_email = "[email protected]",
      platforms="Linux, Darwin",
      ext_modules = [test_module]

test.py

import test
ret = test.hundredstr("hoge");

if ret:
  print ret

testmodule.c

#include <Python.h>

#define DEBUG(...)         \
    do {                   \
        printf("DEBUG: ");   \
        printf(__VA_ARGS__); \
        printf("\n");        \
    } while(0)


typedef struct {
    char* buf;
    size_t buf_size;
    size_t len;
    size_t limit;
} buffer;


buffer*
malloc_buffer(void)
{
    buffer* buf;
    buf = (buffer*)PyMem_Malloc(sizeof(buffer));
    memset(buf, 0, sizeof(buffer));
    return buf;
}

void
dealloc_buffer(buffer* buf)
{
    PyMem_Free(buf);
}

buffer*
create_buffer(size_t buf_size, size_t limit)
{
    buffer* buf;
    buf = malloc_buffer();
    if (!buf) {
        PyErr_NoMemory();
        return NULL;
    }

    buf->buf = PyMem_Malloc(sizeof(char) * buf_size);
    if ( !(buf->buf) ) {
        dealloc_buffer(buf);
        PyErr_NoMemory();
        return NULL;
    }

    buf->buf_size = buf_size;
    buf->len = 0;
    buf->limit = limit;

    DEBUG("new buffer : %p (buf %p, size %d)", buf, buf->buf, (int)buf->buf_size);
    return buf;
}


int
copy_to_buffer(buffer* buf, const char* c, size_t l) {
    char* new_buf;
    size_t new_len = buf->len + l;

    DEBUG("copy_to_buffer : %p", buf->buf);

    if (!buf) {
        PyErr_NoMemory();
        return 1;
    }

    if (new_len >= buf->buf_size) {
        buf->buf_size *= 2;
        if (new_len >= buf->buf_size) {
            buf->buf_size += l + 1;
        }
        if (new_len > buf->limit) {
            buf->buf_size = new_len + buf->limit;
        }

        DEBUG("copy_to_buffer : %p (buf reallocing %p)", buf, buf->buf);
        new_buf = (char*)PyMem_Realloc(buf->buf, buf->buf_size);
        
        if (!new_buf) {
            PyErr_SetString(PyExc_MemoryError, "out of memory");
            free(buf->buf);
            buf->buf = NULL;
            buf->buf_size = buf->len = 0;
            return 1;
        }
        buf->buf = new_buf;
        DEBUG("copy_to_buffer : %p (buf realloced %p)", buf, buf->buf);
    }

    memcpy(buf->buf + buf->len, c, l);
    buf->len += (int)l;

    return 0;
}


PyObject*
test_hundredstr(PyObject* self, PyObject* args) {
    const char* original;
    buffer* buf;
    int i;
    int ret;

    if (!PyArg_ParseTuple(args, "s", &original)) {
        return NULL;
    }
    DEBUG("test_hundredstr : original string -> %s (size %d)", 
          original, (int)strlen(original));

    buf = create_buffer(10, 100);

    for (i = 0; i < 100000; ++i) {
        ret = copy_to_buffer(buf, original, strlen(original));
        if (ret)
            Py_RETURN_NONE;
    }
    ret = copy_to_buffer(buf, "\0", 1);

    return Py_BuildValue("s", buf->buf);
}


/* methods */
static PyMethodDef test_methods[] = {
    {"hundredstr", test_hundredstr, METH_VARARGS,
     "return hundred timed string"},
    {NULL, NULL},
};


void
inittest(void)
{
    Py_InitModule("test", test_methods);
}