自定义大小数组
简单问题描述：
在C或者Cython中，能否创建一个自定义大小的数据类型数组（比如3/5/6/7字节）？
背景：
我在尝试编写一个复杂算法时，发现了一个严重的内存效率问题。这个算法需要存储大量的数据，数据都是连续存放在内存中的（就像一个数组）。这些数据其实就是一长串[通常是]非常大的数字。这个列表/数组中的数字类型是固定的，给定一组数字后，它们的类型是一样的（就像普通的C数组，数组中的所有数字都是同一种类型）。
问题：
有时候，用标准的数据大小来存储每个数字并不高效。通常的标准数据类型有char、short、int、long等……但是如果我用int数组来存储一个只需要3字节就能表示的数据类型，那么每个数字就会浪费1字节的空间。这会导致极大的内存浪费，尤其是当你需要存储数百万个数字时，影响就会非常严重。不幸的是，解决这个算法的办法只有这一种，我认为实现一个自定义数据大小的粗略方法是唯一的解决方案。
我尝试过的：
我尝试使用char数组来完成这个任务，但在不同的0到255的值之间转换成更大的数据类型在大多数情况下都很低效。通常有一种数学方法可以将多个char打包成一个更大的数字，或者将这个更大的数字拆分成单独的char。以下是我在Cython中尝试的一个非常低效的算法：
def to_bytes(long long number, int length):
cdef:
list chars = []
long long m
long long d
for _ in range(length):
m = number % 256
d = number // 256
chars.append(m)
number = d
cdef bytearray binary = bytearray(chars)
binary = binary[::-1]
return binary
def from_bytes(string):
cdef long long d = int(str(string).encode('hex'), 16)
return d
请记住，我并不想对这个算法进行改进，而是想找到一种基本的方法来声明一个特定数据类型的数组，这样我就不需要进行这种转换了。
6 个回答
1
我完全支持使用位集的方法，但要注意对齐的问题。如果你需要频繁地随机访问数据，最好确保你的数据对齐到缓存和CPU的架构上。
另外，我建议你可以考虑另一种方法：
你可以使用像zlib这样的工具，实时解压你需要的数据。如果你预计数据流中会有很多重复的值，这样做可以大大减少输入输出的流量和内存占用。（前提是你对随机访问的需求不是特别高。）这里有一个关于zlib的快速教程。
1
随着处理器处理指令的速度越来越快，我开始想，这种处理方式能有多通用，同时还能在合理的时间内运行。
关于packed位域的问题在于，它们并不是标准的，并且在不同字节序的机器上无法进行读写。我突然想到，小端字节序正好可以解决这个问题……所以我决定利用解决字节序问题的想法，关键是要以小端字节序来存储数据。比如，对于5字节的整数：存储小端字节序的值很简单，只需复制前5个字节；但读取就没那么简单，因为你需要进行符号扩展。
下面的代码可以处理2、3、4和5字节的有符号整数数组：(a) 强制使用小端字节序，(b) 使用packed位域进行比较（见BIT_FIELD）。在给定的情况下，它可以在linux的gcc下编译（64位）。
这段代码有两个假设：
负数是以2的补码或1的补码表示（没有符号和大小）！
结构体的对齐方式为1，可以在任何地址读取/写入，适用于任何大小的结构体。
main函数做了一些测试和计时。它在大数组上运行相同的测试：(a) 'flex'数组，整数长度为2、3、4和5；(b) 简单数组，整数长度为2、4、4和8。在我这台机器上（编译时使用了-O3，最大优化）：
Arrays of 800 million entries -- not using bit-field
With 'flex' arrays of 10.4G bytes: took 20.160 secs: user 16.600 system 3.500
With simple arrays of 13.4G bytes: took 32.580 secs: user 14.680 system 4.910
Arrays of 800 million entries -- using bit-field
With 'flex' arrays of 10.4G bytes: took 22.280 secs: user 18.820 system 3.380
With simple arrays of 13.4G bytes: took 20.450 secs: user 14.450 system 4.620
所以，使用相对通用的代码，特殊长度的整数处理时间更长，但可能没有预期的那么糟糕！！位域版本的速度更慢……我还没时间去研究原因。
所以……我觉得这是可行的。
/*==============================================================================
* 2/3/4/5/... byte "integers" and arrays thereof.
*/
#include <stdint.h>
#include <stdbool.h>
#include <stdlib.h>
#include <stddef.h>
#include <unistd.h>
#include <memory.h>
#include <stdio.h>
#include <sys/times.h>
#include <assert.h>
/*==============================================================================
* General options
*/
#define BIT_FIELD 0 /* use bit-fields (or not) */
#include <endian.h>
#include <byteswap.h>
#if __BYTE_ORDER == __LITTLE_ENDIAN
# define htole16(x) (x)
# define le16toh(x) (x)
# define htole32(x) (x)
# define le32toh(x) (x)
# define htole64(x) (x)
# define le64toh(x) (x)
#else
# define htole16(x) __bswap_16 (x)
# define le16toh(x) __bswap_16 (x)
# define htole32(x) __bswap_32 (x)
# define le32toh(x) __bswap_32 (x)
# define htole64(x) __bswap_64 (x)
# define le64toh(x) __bswap_64 (x)
#endif
typedef int64_t imax_t ;
/*------------------------------------------------------------------------------
* 2 byte integer
*/
#if BIT_FIELD
typedef struct __attribute__((packed)) { int16_t i : 2 * 8 ; } iflex_2b_t ;
#else
typedef struct { int8_t b[2] ; } iflex_2b_t ;
#endif
inline static int16_t
iflex_get_2b(iflex_2b_t item)
{
#if BIT_FIELD
return item.i ;
#else
union
{
int16_t i ;
iflex_2b_t f ;
} x ;
x.f = item ;
return le16toh(x.i) ;
#endif
} ;
inline static iflex_2b_t
iflex_put_2b(int16_t val)
{
#if BIT_FIELD
iflex_2b_t x ;
x.i = val ;
return x ;
#else
union
{
int16_t i ;
iflex_2b_t f ;
} x ;
x.i = htole16(val) ;
return x.f ;
#endif
} ;
/*------------------------------------------------------------------------------
* 3 byte integer
*/
#if BIT_FIELD
typedef struct __attribute__((packed)) { int32_t i : 3 * 8 ; } iflex_3b_t ;
#else
typedef struct { int8_t b[3] ; } iflex_3b_t ;
#endif
inline static int32_t
iflex_get_3b(iflex_3b_t item)
{
#if BIT_FIELD
return item.i ;
#else
union
{
int32_t i ;
int16_t s[2] ;
iflex_2b_t t[2] ;
} x ;
x.t[0] = *((iflex_2b_t*)&item) ;
x.s[1] = htole16(item.b[2]) ;
return le32toh(x.i) ;
#endif
} ;
inline static iflex_3b_t
iflex_put_3b(int32_t val)
{
#if BIT_FIELD
iflex_3b_t x ;
x.i = val ;
return x ;
#else
union
{
int32_t i ;
iflex_3b_t f ;
} x ;
x.i = htole32(val) ;
return x.f ;
#endif
} ;
/*------------------------------------------------------------------------------
* 4 byte integer
*/
#if BIT_FIELD
typedef struct __attribute__((packed)) { int32_t i : 4 * 8 ; } iflex_4b_t ;
#else
typedef struct { int8_t b[4] ; } iflex_4b_t ;
#endif
inline static int32_t
iflex_get_4b(iflex_4b_t item)
{
#if BIT_FIELD
return item.i ;
#else
union
{
int32_t i ;
iflex_4b_t f ;
} x ;
x.f = item ;
return le32toh(x.i) ;
#endif
} ;
inline static iflex_4b_t
iflex_put_4b(int32_t val)
{
#if BIT_FIELD
iflex_4b_t x ;
x.i = val ;
return x ;
#else
union
{
int32_t i ;
iflex_4b_t f ;
} x ;
x.i = htole32((int32_t)val) ;
return x.f ;
#endif
} ;
/*------------------------------------------------------------------------------
* 5 byte integer
*/
#if BIT_FIELD
typedef struct __attribute__((packed)) { int64_t i : 5 * 8 ; } iflex_5b_t ;
#else
typedef struct { int8_t b[5] ; } iflex_5b_t ;
#endif
inline static int64_t
iflex_get_5b(iflex_5b_t item)
{
#if BIT_FIELD
return item.i ;
#else
union
{
int64_t i ;
int32_t s[2] ;
iflex_4b_t t[2] ;
} x ;
x.t[0] = *((iflex_4b_t*)&item) ;
x.s[1] = htole32(item.b[4]) ;
return le64toh(x.i) ;
#endif
} ;
inline static iflex_5b_t
iflex_put_5b(int64_t val)
{
#if BIT_FIELD
iflex_5b_t x ;
x.i = val ;
return x ;
#else
union
{
int64_t i ;
iflex_5b_t f ;
} x ;
x.i = htole64(val) ;
return x.f ;
#endif
} ;
/*------------------------------------------------------------------------------
*
*/
#define alignof(t) __alignof__(t)
/*==============================================================================
* To begin at the beginning...
*/
int
main(int argc, char* argv[])
{
int count = 800 ;
assert(sizeof(iflex_2b_t) == 2) ;
assert(alignof(iflex_2b_t) == 1) ;
assert(sizeof(iflex_3b_t) == 3) ;
assert(alignof(iflex_3b_t) == 1) ;
assert(sizeof(iflex_4b_t) == 4) ;
assert(alignof(iflex_4b_t) == 1) ;
assert(sizeof(iflex_5b_t) == 5) ;
assert(alignof(iflex_5b_t) == 1) ;
clock_t at_start_clock, at_end_clock ;
struct tms at_start_tms, at_end_tms ;
clock_t ticks ;
printf("Arrays of %d million entries -- %susing bit-field\n", count,
BIT_FIELD ? "" : "not ") ;
count *= 1000000 ;
iflex_2b_t* arr2 = malloc(count * sizeof(iflex_2b_t)) ;
iflex_3b_t* arr3 = malloc(count * sizeof(iflex_3b_t)) ;
iflex_4b_t* arr4 = malloc(count * sizeof(iflex_4b_t)) ;
iflex_5b_t* arr5 = malloc(count * sizeof(iflex_5b_t)) ;
size_t bytes = ((size_t)count * (2 + 3 + 4 + 5)) ;
srand(314159) ;
at_start_clock = times(&at_start_tms) ;
for (int i = 0 ; i < count ; i++)
{
imax_t v5, v4, v3, v2, r ;
v2 = (int16_t)(rand() % 0x10000) ;
arr2[i] = iflex_put_2b(v2) ;
v3 = (v2 * 0x100) | ((i & 0xFF) ^ 0x33) ;
arr3[i] = iflex_put_3b(v3) ;
v4 = (v3 * 0x100) | ((i & 0xFF) ^ 0x44) ;
arr4[i] = iflex_put_4b(v4) ;
v5 = (v4 * 0x100) | ((i & 0xFF) ^ 0x55) ;
arr5[i] = iflex_put_5b(v5) ;
r = iflex_get_2b(arr2[i]) ;
assert(r == v2) ;
r = iflex_get_3b(arr3[i]) ;
assert(r == v3) ;
r = iflex_get_4b(arr4[i]) ;
assert(r == v4) ;
r = iflex_get_5b(arr5[i]) ;
assert(r == v5) ;
} ;
for (int i = count - 1 ; i >= 0 ; i--)
{
imax_t v5, v4, v3, v2, r, b ;
v5 = iflex_get_5b(arr5[i]) ;
b = (i & 0xFF) ^ 0x55 ;
assert((v5 & 0xFF) == b) ;
r = (v5 ^ b) / 0x100 ;
v4 = iflex_get_4b(arr4[i]) ;
assert(v4 == r) ;
b = (i & 0xFF) ^ 0x44 ;
assert((v4 & 0xFF) == b) ;
r = (v4 ^ b) / 0x100 ;
v3 = iflex_get_3b(arr3[i]) ;
assert(v3 == r) ;
b = (i & 0xFF) ^ 0x33 ;
assert((v3 & 0xFF) == b) ;
r = (v3 ^ b) / 0x100 ;
v2 = iflex_get_2b(arr2[i]) ;
assert(v2 == r) ;
} ;
at_end_clock = times(&at_end_tms) ;
ticks = sysconf(_SC_CLK_TCK) ;
printf("With 'flex' arrays of %4.1fG bytes: "
"took %5.3f secs: user %5.3f system %5.3f\n",
(double)bytes / (double)(1024 *1024 *1024),
(double)(at_end_clock - at_start_clock) / (double)ticks,
(double)(at_end_tms.tms_utime - at_start_tms.tms_utime) / (double)ticks,
(double)(at_end_tms.tms_stime - at_start_tms.tms_stime) / (double)ticks) ;
free(arr2) ;
free(arr3) ;
free(arr4) ;
free(arr5) ;
int16_t* brr2 = malloc(count * sizeof(int16_t)) ;
int32_t* brr3 = malloc(count * sizeof(int32_t)) ;
int32_t* brr4 = malloc(count * sizeof(int32_t)) ;
int64_t* brr5 = malloc(count * sizeof(int64_t)) ;
bytes = ((size_t)count * (2 + 4 + 4 + 8)) ;
srand(314159) ;
at_start_clock = times(&at_start_tms) ;
for (int i = 0 ; i < count ; i++)
{
imax_t v5, v4, v3, v2, r ;
v2 = (int16_t)(rand() % 0x10000) ;
brr2[i] = v2 ;
v3 = (v2 * 0x100) | ((i & 0xFF) ^ 0x33) ;
brr3[i] = v3 ;
v4 = (v3 * 0x100) | ((i & 0xFF) ^ 0x44) ;
brr4[i] = v4 ;
v5 = (v4 * 0x100) | ((i & 0xFF) ^ 0x55) ;
brr5[i] = v5 ;
r = brr2[i] ;
assert(r == v2) ;
r = brr3[i] ;
assert(r == v3) ;
r = brr4[i] ;
assert(r == v4) ;
r = brr5[i] ;
assert(r == v5) ;
} ;
for (int i = count - 1 ; i >= 0 ; i--)
{
imax_t v5, v4, v3, v2, r, b ;
v5 = brr5[i] ;
b = (i & 0xFF) ^ 0x55 ;
assert((v5 & 0xFF) == b) ;
r = (v5 ^ b) / 0x100 ;
v4 = brr4[i] ;
assert(v4 == r) ;
b = (i & 0xFF) ^ 0x44 ;
assert((v4 & 0xFF) == b) ;
r = (v4 ^ b) / 0x100 ;
v3 = brr3[i] ;
assert(v3 == r) ;
b = (i & 0xFF) ^ 0x33 ;
assert((v3 & 0xFF) == b) ;
r = (v3 ^ b) / 0x100 ;
v2 = brr2[i] ;
assert(v2 == r) ;
} ;
at_end_clock = times(&at_end_tms) ;
printf("With simple arrays of %4.1fG bytes: "
"took %5.3f secs: user %5.3f system %5.3f\n",
(double)bytes / (double)(1024 *1024 *1024),
(double)(at_end_clock - at_start_clock) / (double)ticks,
(double)(at_end_tms.tms_utime - at_start_tms.tms_utime) / (double)ticks,
(double)(at_end_tms.tms_stime - at_start_tms.tms_stime) / (double)ticks) ;
free(brr2) ;
free(brr3) ;
free(brr4) ;
free(brr5) ;
return 0 ;
} ;
1
你可以使用一种叫做“紧凑位域”的东西。在GCC中，它的写法大概是这样的：
typedef struct __attribute__((__packed__)) {
int x : 24;
} int24;
对于一个名为 x 的24位整数，x.x 的表现就像一个24位的整数一样。你可以创建一个这样的数组，而且它不会有多余的空隙。不过要注意，这种方式的速度会比普通整数慢，因为数据不会对齐，而且我觉得没有专门的指令来读取24位的数据。编译器在每次读取和存储时都需要生成额外的代码。
1
在C语言中，你可以定义一个自定义的数据类型，这样就能处理任意字节大小的复杂情况：
typedef struct 3byte { char x[3]; } 3byte;
这样一来，你就可以做很多方便的事情，比如按值传递、获取正确的size_t类型，以及创建这种类型的数组。
1
我觉得重要的问题是你是否需要同时访问所有数据。
如果你只需要同时访问一部分数据
如果你一次只需要访问一个数组，那么可以考虑使用NumPy数组，数据类型选择uint8，宽度根据需要来定。当你需要处理数据时，可以把压缩的数据展开（这里是把3个字节的数字展开成uint32）：
import numpy as np
# in this example `compressed` is a Nx3 array of octets (`uint8`)
expanded = np.empty((compressed.shape[0], 4))
expanded[:,:3] = compressed
expanded[:, 3] = 0
expanded = expanded.view('uint32').reshape(-1)
然后操作是在expanded上进行的，它是一个包含N个uint32值的一维向量。
完成后，数据可以保存回去：
# recompress
compressed[:] = expanded.view('uint8').reshape(-1,4)[:,:3]
在我的机器上，处理每个方向的时间大约是每个元素8纳秒（使用Python）。使用Cython可能不会带来太大的性能提升，因为几乎所有的时间都花在了在NumPy的某个深处复制数据。
这是一种一次性的高成本，但如果你打算至少访问每个元素一次，支付这一次性成本可能比每次操作都付出类似的成本要划算。
当然，C语言也可以采用相同的方法：
#include <stdlib.h>
#include <stdio.h>
#include <time.h>
#include <sys/resource.h>
#define NUMITEMS 10000000
int main(void)
{
uint32_t *expanded;
uint8_t * cmpressed, *exp_as_octets;
struct rusage ru0, ru1;
uint8_t *ep, *cp, *end;
double time_delta;
// create some compressed data
cmpressed = (uint8_t *)malloc(NUMITEMS * 3);
getrusage(RUSAGE_SELF, &ru0);
// allocate the buffer and copy the data
exp_as_octets = (uint8_t *)malloc(NUMITEMS * 4);
end = exp_as_octets + NUMITEMS * 4;
ep = exp_as_octets;
cp = cmpressed;
while (ep < end)
{
// copy three octets out of four
*ep++ = *cp++;
*ep++ = *cp++;
*ep++ = *cp++;
*ep++ = 0;
}
expanded = (uint32_t *)exp_as_octets;
getrusage(RUSAGE_SELF, &ru1);
printf("Uncompress\n");
time_delta = ru1.ru_utime.tv_sec + ru1.ru_utime.tv_usec * 1e-6
- ru0.ru_utime.tv_sec - ru0.ru_utime.tv_usec * 1e-6;
printf("User: %.6lf seconds, %.2lf nanoseconds per element", time_delta, 1e9 * time_delta / NUMITEMS);
time_delta = ru1.ru_stime.tv_sec + ru1.ru_stime.tv_usec * 1e-6
- ru0.ru_stime.tv_sec - ru0.ru_stime.tv_usec * 1e-6;
printf("System: %.6lf seconds, %.2lf nanoseconds per element", time_delta, 1e9 * time_delta / NUMITEMS);
getrusage(RUSAGE_SELF, &ru0);
// compress back
ep = exp_as_octets;
cp = cmpressed;
while (ep < end)
{
*cp++ = *ep++;
*cp++ = *ep++;
*cp++ = *ep++;
ep++;
}
getrusage(RUSAGE_SELF, &ru1);
printf("Compress\n");
time_delta = ru1.ru_utime.tv_sec + ru1.ru_utime.tv_usec * 1e-6
- ru0.ru_utime.tv_sec - ru0.ru_utime.tv_usec * 1e-6;
printf("User: %.6lf seconds, %.2lf nanoseconds per element", time_delta, 1e9 * time_delta / NUMITEMS);
time_delta = ru1.ru_stime.tv_sec + ru1.ru_stime.tv_usec * 1e-6
- ru0.ru_stime.tv_sec - ru0.ru_stime.tv_usec * 1e-6;
printf("System: %.6lf seconds, %.2lf nanoseconds per element", time_delta, 1e9 * time_delta / NUMITEMS);
}
这段代码的输出是：
Uncompress
User: 0.022650 seconds, 2.27 nanoseconds per element
System: 0.016171 seconds, 1.62 nanoseconds per element
Compress
User: 0.011698 seconds, 1.17 nanoseconds per element
System: 0.000018 seconds, 0.00 nanoseconds per element
代码是用gcc -Ofast编译的，速度可能比较接近最佳。系统时间主要花在malloc上。看起来速度相当快，因为我们以2-3 GB/s的速度读取内存。（这也意味着，虽然让代码支持多线程很简单，但可能并不会带来太大的速度提升。）
如果你想要最佳性能，就需要为每种数据宽度单独编写压缩/解压缩的例程。（我并不保证上面的C代码在任何机器上都是最快的，我没有查看机器代码。）
如果你需要随机访问不同的值
如果你只需要在这里访问一个值，在那里访问另一个值，Python不会提供任何合理快速的方法，因为数组查找的开销很大。
在这种情况下，我建议你创建C语言的例程来获取和存放数据。可以参考technosaurus的回答。有很多技巧，但对齐问题是无法避免的。
在读取奇数大小的数组时，一个有用的技巧可能是（这里是从一个字节数组compressed中读取3个字节到一个uint32_t的value中）：
value = (uint32_t *)&compressed[3 * n] & 0x00ffffff;
然后其他人会处理可能的对齐问题，最后会有一个字节的垃圾数据。不幸的是，这在写入值时无法使用。而且——再次强调——这可能比其他任何替代方案快，也可能慢。