/****************************************************************************** * page_alloc.c * * Simple buddy heap allocator for Xen. * * Copyright (c) 2002-2004 K A Fraser * * This program is free software; you can redistribute it and/or modify * it under the terms of the GNU General Public License as published by * the Free Software Foundation; either version 2 of the License, or * (at your option) any later version. * * This program is distributed in the hope that it will be useful, * but WITHOUT ANY WARRANTY; without even the implied warranty of * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the * GNU General Public License for more details. * * You should have received a copy of the GNU General Public License * along with this program; if not, write to the Free Software * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */ #include #include #include #include #include #include #include #include #include #include #include #include #include /* * Comma-separated list of hexadecimal page numbers containing bad bytes. * e.g. 'badpage=0x3f45,0x8a321'. */ static char opt_badpage[100] = ""; string_param("badpage", opt_badpage); #define round_pgdown(_p) ((_p)&PAGE_MASK) #define round_pgup(_p) (((_p)+(PAGE_SIZE-1))&PAGE_MASK) static spinlock_t page_scrub_lock = SPIN_LOCK_UNLOCKED; LIST_HEAD(page_scrub_list); /********************* * ALLOCATION BITMAP * One bit per page of memory. Bit set => page is allocated. */ static unsigned long bitmap_size; /* in bytes */ static unsigned long *alloc_bitmap; #define PAGES_PER_MAPWORD (sizeof(unsigned long) * 8) #define allocated_in_map(_pn) \ ( !! (alloc_bitmap[(_pn)/PAGES_PER_MAPWORD] & \ (1UL<<((_pn)&(PAGES_PER_MAPWORD-1)))) ) /* * Hint regarding bitwise arithmetic in map_{alloc,free}: * -(1<= n. * (1<> PAGE_SHIFT, (pe - ps) >> PAGE_SHIFT); /* Check new pages against the bad-page list. */ p = opt_badpage; while ( *p != '\0' ) { bad_pfn = simple_strtoul(p, &p, 0); if ( *p == ',' ) p++; else if ( *p != '\0' ) break; if ( (bad_pfn < (bitmap_size*8)) && !allocated_in_map(bad_pfn) ) { printk("Marking page %lx as bad\n", bad_pfn); map_alloc(bad_pfn, 1); } } } unsigned long alloc_boot_pages(unsigned long nr_pfns, unsigned long pfn_align) { unsigned long pg, i; for ( pg = 0; (pg + nr_pfns) < (bitmap_size*8); pg += pfn_align ) { for ( i = 0; i < nr_pfns; i++ ) if ( allocated_in_map(pg + i) ) break; if ( i == nr_pfns ) { map_alloc(pg, nr_pfns); return pg; } } return 0; } /************************* * BINARY BUDDY ALLOCATOR */ #define MEMZONE_XEN 0 #define MEMZONE_DOM 1 #define NR_ZONES 2 /* Up to 2^20 pages can be allocated at once. */ #define MAX_ORDER 20 static struct list_head heap[NR_ZONES][MAX_ORDER+1]; static unsigned long avail[NR_ZONES]; static spinlock_t heap_lock = SPIN_LOCK_UNLOCKED; void end_boot_allocator(void) { unsigned long i, j; int curr_free = 0, next_free = 0; memset(avail, 0, sizeof(avail)); for ( i = 0; i < NR_ZONES; i++ ) for ( j = 0; j <= MAX_ORDER; j++ ) INIT_LIST_HEAD(&heap[i][j]); /* Pages that are free now go to the domain sub-allocator. */ for ( i = 0; i < max_page; i++ ) { curr_free = next_free; next_free = !allocated_in_map(i+1); if ( next_free ) map_alloc(i+1, 1); /* prevent merging in free_heap_pages() */ if ( curr_free ) free_heap_
from io import BytesIO
import sys
from netlib import wsgi
from netlib.http import Headers


def tflow():
    headers = Headers(test=b"value")
    req = wsgi.Request("http", "GET", "/", "HTTP/1.1", headers, "")
    return wsgi.Flow(("127.0.0.1", 8888), req)


class ExampleApp:

    def __init__(self):
        self.called = False

    def __call__(self, environ, start_response):
        self.called = True
        status = '200 OK'
        response_headers = [('Content-type', 'text/plain')]
        start_response(status, response_headers)
        return [b'Hello', b' world!\n']


class TestWSGI:

    def test_make_environ(self):
        w = wsgi.WSGIAdaptor(None, "foo", 80, "version")
        tf = tflow()
        assert w.make_environ(tf, None)

        tf.request.path = "/foo?bar=voing"
        r = w.make_environ(tf, None)
        assert r["QUERY_STRING"] == "bar=voing"

    def test_serve(self):
        ta = ExampleApp()
        w = wsgi.WSGIAdaptor(ta, "foo", 80, "version")
        f = tflow()
        f.request.host = "foo"
        f.request.port = 80

        wfile = BytesIO()
        err = w.serve(f, wfile)
        assert ta.called
        assert not err

        val = wfile.getvalue()
        assert b"Hello world" in val
        assert b"Server:" in val

    def _serve(self, app):
        w = wsgi.WSGIAdaptor(app, "foo", 80, "version")
        f = tflow()
        f.request.host = "foo"
        f.request.port = 80
        wfile = BytesIO()
        w.serve(f, wfile)
        return wfile.getvalue()

    def test_serve_empty_body(self):
        def app(environ, start_response):
            status = '200 OK'
            response_headers = [('Foo', 'bar')]
            start_response(status, response_headers)
            return []
        assert self._serve(app)

    def test_serve_double_start(self):
        def app(environ, start_response):
            try:
                raise ValueError("foo")
            except:
                sys.exc_info()
            status = '200 OK'
            response_headers = [('Content-type', 'text/plain')]
            start_response(status, response_headers)
            start_response(status, response_headers)
        assert b"Internal Server Error" in self._serve(app)

    def test_serve_single_err(self):
        def app(environ, start_response):
            try:
                raise ValueError("foo")
            except:
                ei = sys.exc_info()
            status = '200 OK'
            response_headers = [('Content-type', 'text/plain')]
            start_response(status, response_headers, ei)
            yield b""
        assert b"Internal Server Error" in self._serve(app)

    def test_serve_double_err(self):
        def app(environ, start_response):
            try:
                raise ValueError("foo")
            except:
                ei = sys.exc_info()
            status = '200 OK'
            response_headers = [('Content-type', 'text/plain')]
            start_response(status, response_headers)
            yield b"aaa"
            start_response(status, response_headers, ei)
            yield b"bbb"
        assert b"Internal Server Error" in self._serve(app)
generated by cgit v1.2.3 (git 2.25.1) at 2025-06-21 03:08:26 +0000