fix: remove memory allocation in hash maps (#298)

* add bpf_get_smp_processor_id

* add xdp related functions

* add helpers and map in maps

* fix map access memory alloc

* fix name error

* remove duplicate assign

* fix hash map impl

* Update map_common_def.hpp

runtime/src/bpf_map/map_common_def.hpp

@@ -23,20 +23,7 @@ using bytes_vec = boost::interprocess::vector<uint8_t, bytes_vec_allocator>;
 template <class T>
 static inline T ensure_on_current_cpu(std::function<T(int cpu)> func)
 {
-	static thread_local int currcpu = -1;
-	if (currcpu == sched_getcpu()) {
-		return func(currcpu);
-	}
-	cpu_set_t orig, set;
-	CPU_ZERO(&orig);
-	CPU_ZERO(&set);
-	sched_getaffinity(0, sizeof(orig), &orig);
-	currcpu = sched_getcpu();
-	CPU_SET(currcpu, &set);
-	sched_setaffinity(0, sizeof(set), &set);
-	T ret = func(currcpu);
-	sched_setaffinity(0, sizeof(orig), &orig);
-	return ret;
+	return func(sched_getcpu());
 }
 
 template <class T>

runtime/src/bpf_map/userspace/hash_map.cpp

@@ -25,9 +25,8 @@ hash_map_impl::hash_map_impl(managed_shared_memory &memory, uint32_t key_size,
 void *hash_map_impl::elem_lookup(const void *key)
 {
 	SPDLOG_TRACE("Peform elem lookup of hash map");
-	// Allocate as a local variable to make
-	//  it thread safe, since we use sharable lock
-	bytes_vec key_vec = this->key_vec;
+	// Since we use lock here, we don't need to allocate key_vec and
+	// value_vec
 	key_vec.assign((uint8_t *)key, (uint8_t *)key + _key_size);
 	if (auto itr = map_impl.find(key_vec); itr != map_impl.end()) {
 		SPDLOG_TRACE("Exit elem lookup of hash map");
@@ -42,21 +41,14 @@ void *hash_map_impl::elem_lookup(const void *key)
 long hash_map_impl::elem_update(const void *key, const void *value,
 				uint64_t flags)
 {
-	bytes_vec key_vec = this->key_vec;
-	bytes_vec value_vec = this->value_vec;
 	key_vec.assign((uint8_t *)key, (uint8_t *)key + _key_size);
 	value_vec.assign((uint8_t *)value, (uint8_t *)value + _value_size);
-	if (auto itr = map_impl.find(key_vec); itr != map_impl.end()) {
-		itr->second = value_vec;
-	} else {
-		map_impl.insert(bi_map_value_ty(key_vec, value_vec));
-	}
+	map_impl.insert_or_assign(key_vec, value_vec);
 	return 0;
 }
 
 long hash_map_impl::elem_delete(const void *key)
 {
-	bytes_vec key_vec = this->key_vec;
 	key_vec.assign((uint8_t *)key, (uint8_t *)key + _key_size);
 	map_impl.erase(key_vec);
 	return 0;
@@ -75,9 +67,8 @@ int hash_map_impl::map_get_next_key(const void *key, void *next_key)
 			  (uint8_t *)next_key);
 		return 0;
 	}
-	// No need to be allocated at shm. Allocate as a local variable to make
-	// it thread safe, since we use sharable lock
-	bytes_vec key_vec = this->key_vec;
+	// Since we use lock here, we don't need to allocate key_vec and
+	// value_vec
 	key_vec.assign((uint8_t *)key, (uint8_t *)key + _key_size);
 
 	auto itr = map_impl.find(key_vec);

runtime/src/bpf_map/userspace/hash_map.hpp

@@ -31,6 +31,7 @@ class hash_map_impl {
 	uint32_t _key_size;
 	uint32_t _value_size;
 
+	// buffers used to access the key and value in hash map
 	bytes_vec key_vec;
 	bytes_vec value_vec;
 

runtime/src/bpf_map/userspace/map_in_maps.hpp

@@ -15,13 +15,18 @@ namespace bpftime
 // implementation of array map
 class array_map_of_maps_impl : public array_map_impl {
     public:
-	array_map_of_maps_impl(boost::interprocess::managed_shared_memory &memory, uint32_t max_entries) : array_map_impl(memory, sizeof(int), max_entries){
-			   }
+	array_map_of_maps_impl(
+		boost::interprocess::managed_shared_memory &memory,
+		uint32_t max_entries)
+		: array_map_impl(memory, sizeof(int), max_entries)
+	{
+	}
 	// TODO: add verify the correctness of the key
-	void *elem_lookup(const void *key) {
+	void *elem_lookup(const void *key)
+	{
 		auto key_val = array_map_impl::elem_lookup(key);
 		int map_id = *(int *)key_val;
-		return (void*)((u_int64_t)map_id << 32);
+		return (void *)((u_int64_t)map_id << 32);
 	}
 };
 

runtime/src/bpf_map/userspace/per_cpu_array_map.cpp

@@ -71,9 +71,6 @@ long per_cpu_array_map_impl::elem_delete(const void *key)
 	errno = ENOTSUP;
 	SPDLOG_ERROR("Deleting of per cpu array is not supported");
 	return -1;
-	// return ensure_on_current_cpu<long>([&](int cpu) {
-
-	// });
 }
 
 int per_cpu_array_map_impl::map_get_next_key(const void *key,

runtime/src/bpf_map/userspace/per_cpu_hash_map.cpp

@@ -25,74 +25,78 @@ per_cpu_hash_map_impl::per_cpu_hash_map_impl(
 	uint32_t value_size, int ncpu)
 	: impl(memory.get_segment_manager()), key_size(key_size),
 	  value_size(value_size), ncpu(ncpu),
-	  key_template(key_size, memory.get_segment_manager()),
 	  value_template(value_size * ncpu, memory.get_segment_manager()),
-	  single_value_template(value_size, memory.get_segment_manager())
+	  key_templates(memory.get_segment_manager()),
+	  single_value_templates(memory.get_segment_manager())
 {
 	SPDLOG_DEBUG(
 		"Initializing per cpu hash, key size {}, value size {}, ncpu {}",
 		key_size, value_size, ncpu);
+	for (int i = 0; i < ncpu; i++) {
+		bytes_vec key_vec(key_size, memory.get_segment_manager());
+		key_templates.push_back(key_vec);
+		bytes_vec value_vec(value_size, memory.get_segment_manager());
+		single_value_templates.push_back(value_vec);
+	}
 }
 
 void *per_cpu_hash_map_impl::elem_lookup(const void *key)
 {
-	return ensure_on_current_cpu<void *>([&](int cpu) -> void * {
-		SPDLOG_DEBUG("Run per cpu hash lookup at cpu {}", cpu);
-		if (key == nullptr) {
-			errno = ENOENT;
-			return nullptr;
-		}
-		bytes_vec key_vec = this->key_template;
-		key_vec.assign((uint8_t *)key, (uint8_t *)key + key_size);
-		if (auto itr = impl.find(key_vec); itr != impl.end()) {
-			SPDLOG_TRACE("Exit elem lookup of hash map");
-			return &itr->second[value_size * cpu];
-		} else {
-			SPDLOG_TRACE("Exit elem lookup of hash map");
-			errno = ENOENT;
-			return nullptr;
-		}
-	});
+	int cpu = sched_getcpu();
+	SPDLOG_DEBUG("Run per cpu hash lookup at cpu {}", cpu);
+	if (key == nullptr) {
+		errno = ENOENT;
+		return nullptr;
+	}
+	bytes_vec &key_vec = this->key_templates[cpu];
+	key_vec.assign((uint8_t *)key, (uint8_t *)key + key_size);
+	if (auto itr = impl.find(key_vec); itr != impl.end()) {
+		SPDLOG_TRACE("Exit elem lookup of hash map");
+		return &itr->second[value_size * cpu];
+	} else {
+		SPDLOG_TRACE("Exit elem lookup of hash map");
+		errno = ENOENT;
+		return nullptr;
+	}
 }
 
 long per_cpu_hash_map_impl::elem_update(const void *key, const void *value,
 					uint64_t flags)
 {
-	SPDLOG_DEBUG("Per cpu update, key {}, value {}",(const char*)key,*(long*)value);
-	return ensure_on_current_cpu<long>([&](int cpu) -> long {
-		SPDLOG_DEBUG("Run per cpu hash update at cpu {}", cpu);
-		bytes_vec key_vec = this->key_template;
-		bytes_vec value_vec = this->single_value_template;
-		key_vec.assign((uint8_t *)key, (uint8_t *)key + key_size);
-		value_vec.assign((uint8_t *)value,
-				 (uint8_t *)value + value_size);
-		if (auto itr = impl.find(key_vec); itr != impl.end()) {
-			std::copy(value_vec.begin(), value_vec.end(),
-				  itr->second.begin() + cpu * value_size);
-		} else {
-			bytes_vec full_value_vec = this->value_template;
-			std::copy(value_vec.begin(), value_vec.end(),
-				  full_value_vec.begin() + cpu * value_size);
+	int cpu = sched_getcpu();
+	SPDLOG_DEBUG("Per cpu update, key {}, value {}", (const char *)key,
+		     *(long *)value);
 
-			impl.insert(bi_map_value_ty(key_vec, full_value_vec));
-		}
+	SPDLOG_DEBUG("Run per cpu hash update at cpu {}", cpu);
+	bytes_vec &key_vec = this->key_templates[cpu];
+	bytes_vec &value_vec = this->single_value_templates[cpu];
+	key_vec.assign((uint8_t *)key, (uint8_t *)key + key_size);
+	value_vec.assign((uint8_t *)value, (uint8_t *)value + value_size);
+	if (auto itr = impl.find(key_vec); itr != impl.end()) {
+		std::copy(value_vec.begin(), value_vec.end(),
+			  itr->second.begin() + cpu * value_size);
+	} else {
+		bytes_vec full_value_vec = this->value_template;
+		std::copy(value_vec.begin(), value_vec.end(),
+			  full_value_vec.begin() + cpu * value_size);
 
-		return 0;
-	});
+		impl.insert(bi_map_value_ty(key_vec, full_value_vec));
+	}
+
+	return 0;
 }
 
 long per_cpu_hash_map_impl::elem_delete(const void *key)
 {
-	return ensure_on_current_cpu<long>([&](int cpu) -> long {
-		SPDLOG_DEBUG("Run per cpu hash delete at cpu {}", cpu);
-		bytes_vec key_vec = this->key_template;
-		key_vec.assign((uint8_t *)key, (uint8_t *)key + key_size);
-		if (auto itr = impl.find(key_vec); itr != impl.end()) {
-			std::fill(itr->second.begin(),
-				  itr->second.begin() + cpu * value_size, 0);
-		}
-		return 0;
-	});
+	int cpu = sched_getcpu();
+	SPDLOG_DEBUG("Run per cpu hash delete at cpu {}", cpu);
+	bytes_vec &key_vec = this->key_templates[cpu];
+	key_vec.assign((uint8_t *)key, (uint8_t *)key + key_size);
+	if (auto itr = impl.find(key_vec); itr != impl.end()) {
+		std::fill(itr->second.begin(),
+			  itr->second.begin() + cpu * value_size, 0);
+	}
+	return 0;
 }
 
 int per_cpu_hash_map_impl::map_get_next_key(const void *key, void *next_key)
@@ -110,7 +114,7 @@ int per_cpu_hash_map_impl::map_get_next_key(const void *key, void *next_key)
 	}
 	// No need to be allocated at shm. Allocate as a local variable to make
 	// it thread safe, since we use sharable lock
-	bytes_vec key_vec = this->key_template;
+	bytes_vec key_vec = this->key_templates[0];
 	key_vec.assign((uint8_t *)key, (uint8_t *)key + key_size);
 
 	auto itr = impl.find(key_vec);
@@ -134,10 +138,12 @@ void *per_cpu_hash_map_impl::elem_lookup_userspace(const void *key)
 		errno = ENOENT;
 		return nullptr;
 	}
-	bytes_vec key_vec = this->key_template;
+	bytes_vec key_vec = this->key_templates[0];
 	key_vec.assign((uint8_t *)key, (uint8_t *)key + key_size);
 	if (auto itr = impl.find(key_vec); itr != impl.end()) {
-		SPDLOG_TRACE("Exit elem lookup of hash map: {}",spdlog::to_hex(itr->second.begin(),itr->second.end()));
+		SPDLOG_TRACE("Exit elem lookup of hash map: {}",
+			     spdlog::to_hex(itr->second.begin(),
+					    itr->second.end()));
 		return &itr->second[0];
 	} else {
 		SPDLOG_TRACE("Exit elem lookup of hash map");
@@ -150,7 +156,7 @@ long per_cpu_hash_map_impl::elem_update_userspace(const void *key,
 						  const void *value,
 						  uint64_t flags)
 {
-	bytes_vec key_vec = this->key_template;
+	bytes_vec key_vec = this->key_templates[0];
 	bytes_vec value_vec = this->value_template;
 	key_vec.assign((uint8_t *)key, (uint8_t *)key + key_size);
 	value_vec.assign((uint8_t *)value,
@@ -165,7 +171,7 @@ long per_cpu_hash_map_impl::elem_update_userspace(const void *key,
 }
 long per_cpu_hash_map_impl::elem_delete_userspace(const void *key)
 {
-	bytes_vec key_vec = this->key_template;
+	bytes_vec key_vec = this->key_templates[0];
 	key_vec.assign((uint8_t *)key, (uint8_t *)key + key_size);
 	impl.erase(key_vec);
 	return 0;

runtime/src/bpf_map/userspace/per_cpu_hash_map.hpp

@@ -26,13 +26,21 @@ class per_cpu_hash_map_impl {
 		boost::unordered_map<bytes_vec, bytes_vec, bytes_vec_hasher,
 				     std::equal_to<bytes_vec>, bi_map_allocator>;
 
+	using shm_hash_map_vec_allocator = boost::interprocess::allocator<
+		shm_hash_map, boost::interprocess::managed_shared_memory::segment_manager>;
+	using shm_hash_map_vec = boost::interprocess::vector<shm_hash_map, shm_hash_map_vec_allocator>;
+
+	using bytes_vec_vec_allocator = boost::interprocess::allocator<
+		bytes_vec, boost::interprocess::managed_shared_memory::segment_manager>;
+	using bytes_vec_vec = boost::interprocess::vector<bytes_vec, bytes_vec_vec_allocator>;
+
 	shm_hash_map impl;
 	uint32_t key_size;
 	uint32_t value_size;
 	int ncpu;
 
-	bytes_vec key_template, value_template, single_value_template;
-
+	bytes_vec value_template;
+	bytes_vec_vec key_templates, single_value_templates;
     public:
 	const static bool should_lock = false;
 

runtime/src/bpf_map/userspace/prog_array.cpp

@@ -98,6 +98,7 @@ void *prog_array_map_impl::elem_lookup(const void *key)
 	current_thread_lookup_val = fd;
 	return &current_thread_lookup_val;
 }
+
 long prog_array_map_impl::elem_update(const void *key, const void *value,
 				      uint64_t flags)
 {